diff --git a/crates/dbx-core/src/database_export.rs b/crates/dbx-core/src/database_export.rs index 824e42fb53..12f2b27660 100644 --- a/crates/dbx-core/src/database_export.rs +++ b/crates/dbx-core/src/database_export.rs @@ -809,7 +809,7 @@ pub(crate) fn is_mysql_spatial_export_type(column_type: &str) -> bool { /// Database exports encode MySQL spatial cells as `DBX_WKB::` while /// reading them. Keeping this marker internal lets the normal JSON row shape /// and all non-export query paths continue to expose readable WKT values. -fn format_mysql_spatial_export_literal( +pub(crate) fn format_mysql_spatial_export_literal( value: &Value, database_type: Option, column_type: Option<&str>, @@ -1487,7 +1487,7 @@ fn database_export_select_list(columns: &[String], column_types: &[Option], diff --git a/crates/dbx-core/src/transfer.rs b/crates/dbx-core/src/transfer.rs index ce3235c5f0..4dc2e739e0 100644 --- a/crates/dbx-core/src/transfer.rs +++ b/crates/dbx-core/src/transfer.rs @@ -2972,7 +2972,7 @@ pub fn generate_insert_typed( } let template = InsertSqlTemplate::new(columns, table, schema, db_type, catalog, false); - let value_rows = value_rows_sql(rows, column_types, db_type); + let value_rows = value_rows_sql(rows, column_types, db_type, false); template.build(&value_rows) } @@ -3054,12 +3054,20 @@ fn value_rows_sql( rows: &[Vec], column_types: &[Option], db_type: &DatabaseType, + mysql_spatial_markers: bool, ) -> Vec { let mut out = Vec::with_capacity(rows.len()); for row in rows { let mut vals = Vec::with_capacity(row.len()); for (index, v) in row.iter().enumerate() { - vals.push(escape_value_typed(v, db_type, column_types.get(index).and_then(|value| value.as_deref()))); + let column_type = column_types.get(index).and_then(|value| value.as_deref()); + let value = if mysql_spatial_markers { + crate::database_export::format_mysql_spatial_export_literal(v, Some(*db_type), column_type) + } else { + None + } + .unwrap_or_else(|| escape_value_typed(v, db_type, column_type)); + vals.push(value); } out.push(format!("({})", vals.join(", "))); } @@ -3087,7 +3095,18 @@ pub fn generate_upsert_typed( pk_columns: &[String], catalog: Option<&str>, ) -> String { - generate_upsert_typed_for_transfer(columns, column_types, rows, table, schema, db_type, pk_columns, catalog, false) + generate_upsert_typed_for_transfer( + columns, + column_types, + rows, + table, + schema, + db_type, + pk_columns, + catalog, + false, + false, + ) } #[allow(clippy::too_many_arguments)] @@ -3101,6 +3120,7 @@ fn generate_upsert_typed_for_transfer( pk_columns: &[String], catalog: Option<&str>, overrides_postgres_system_values: bool, + mysql_spatial_markers: bool, ) -> String { if rows.is_empty() || pk_columns.is_empty() { return String::new(); @@ -3109,7 +3129,7 @@ fn generate_upsert_typed_for_transfer( let full_table = qualified_table(table, schema, db_type, catalog); let col_list = columns.iter().map(|c| quote_identifier(c, db_type)).collect::>().join(", "); - let value_rows = value_rows_sql(rows, column_types, db_type); + let value_rows = value_rows_sql(rows, column_types, db_type, mysql_spatial_markers); let mut non_pk_columns = Vec::with_capacity(columns.len().saturating_sub(pk_columns.len())); for c in columns { @@ -3243,7 +3263,10 @@ fn generate_upsert_typed_for_transfer( sql.push_str(&format!("\nWHEN NOT MATCHED THEN INSERT ({insert_cols}) VALUES ({insert_vals})")); sql } - _ => generate_insert_typed(columns, column_types, rows, table, schema, db_type, catalog), + _ => { + let template = InsertSqlTemplate::new(columns, table, schema, db_type, catalog, false); + template.build(&value_rows_sql(rows, column_types, db_type, mysql_spatial_markers)) + } } } @@ -3301,6 +3324,22 @@ fn rewrite_transfer_source_table_ddl( } } +fn mysql_spatial_transfer_select_sql( + sql: String, + columns: &[String], + column_types: &[Option], + source_db_type: &DatabaseType, + target_db_type: &DatabaseType, +) -> (String, bool) { + let has_spatial_columns = column_types + .iter() + .any(|column_type| column_type.as_deref().is_some_and(crate::database_export::is_mysql_spatial_export_type)); + if !matches!((source_db_type, target_db_type), (DatabaseType::Mysql, DatabaseType::Mysql)) || !has_spatial_columns { + return (sql, false); + } + (crate::database_export::replace_database_export_select_list(sql, columns, column_types, source_db_type), true) +} + #[allow(clippy::too_many_arguments)] fn generate_transfer_write_sql( mode: &TransferMode, @@ -3313,6 +3352,7 @@ fn generate_transfer_write_sql( pk_columns: &[String], catalog: Option<&str>, overrides_postgres_system_values: bool, + mysql_spatial_markers: bool, ) -> String { match mode { TransferMode::Upsert => generate_upsert_typed_for_transfer( @@ -3325,6 +3365,7 @@ fn generate_transfer_write_sql( pk_columns, catalog, overrides_postgres_system_values, + mysql_spatial_markers, ), _ => { if rows.is_empty() { @@ -3332,7 +3373,7 @@ fn generate_transfer_write_sql( } let template = InsertSqlTemplate::new(columns, table, schema, db_type, catalog, overrides_postgres_system_values); - template.build(&value_rows_sql(rows, column_types, db_type)) + template.build(&value_rows_sql(rows, column_types, db_type, mysql_spatial_markers)) } } } @@ -3358,6 +3399,7 @@ pub(crate) fn generate_insert_typed_sql_batches( catalog, limits, false, + false, ) } @@ -3372,6 +3414,7 @@ fn generate_insert_typed_sql_batches_for_transfer( catalog: Option<&str>, limits: SqlBatchLimits, overrides_postgres_system_values: bool, + mysql_spatial_markers: bool, ) -> Result, String> { if rows.is_empty() { return Ok(Vec::new()); @@ -3385,7 +3428,7 @@ fn generate_insert_typed_sql_batches_for_transfer( let target_sql_bytes = limits.target_sql_bytes.max(1); let batch_sql_bytes = limits.hard_sql_bytes.map_or(target_sql_bytes, |hard| target_sql_bytes.min(hard)); let template = InsertSqlTemplate::new(columns, table, schema, db_type, catalog, overrides_postgres_system_values); - let value_rows = value_rows_sql(rows, column_types, db_type); + let value_rows = value_rows_sql(rows, column_types, db_type, mysql_spatial_markers); let value_row_bytes = value_rows.iter().map(|row| sql_text_bytes(row, db_type)).collect::>(); let mut statements = Vec::new(); let mut start = 0usize; @@ -3434,6 +3477,7 @@ fn generate_transfer_write_sql_batches( pk_columns: &[String], catalog: Option<&str>, overrides_postgres_system_values: bool, + mysql_spatial_markers: bool, ) -> Result, String> { if rows.is_empty() { return Ok(Vec::new()); @@ -3450,6 +3494,7 @@ fn generate_transfer_write_sql_batches( catalog, SqlBatchLimits::for_database(db_type, max_transfer_write_rows(db_type, mode)), overrides_postgres_system_values, + mysql_spatial_markers, )? .into_iter() .map(|(sql, _)| sql) @@ -3477,6 +3522,7 @@ fn generate_transfer_write_sql_batches( pk_columns, catalog, overrides_postgres_system_values, + mysql_spatial_markers, ); while end < rows.len() && end - start < max_rows { @@ -3491,6 +3537,7 @@ fn generate_transfer_write_sql_batches( pk_columns, catalog, overrides_postgres_system_values, + mysql_spatial_markers, ); if candidate.len() > max_sql_bytes && !accepted.is_empty() { break; @@ -5997,6 +6044,7 @@ where &[], request.target_catalog.as_deref(), false, + false, )?; for (statement_index, batch_sql) in write_statements.iter().enumerate() { execute_on_pool(state, target_pool_key, batch_sql).await.map_err(|e| { @@ -6415,7 +6463,6 @@ where && selected_columns_include_identity_columns(&col_names, &target_columns); let overrides_postgres_system_values = matches!(target_db_type, DatabaseType::Postgres) && selected_columns_include_postgres_generated_always_identity_columns(&col_names, &target_columns); - // Transfer data in batches let batch_size = if request.batch_size == 0 { 1000 } else { request.batch_size }; let mut offset: u64 = 0; @@ -6436,6 +6483,8 @@ where &primary_key_columns, request.source_catalog.as_deref(), ); + let (sql, mysql_spatial_markers) = + mysql_spatial_transfer_select_sql(sql, &col_names, &col_types, source_db_type, target_db_type); let result = execute_on_pool(state, source_pool_key, &sql).await?; let row_count = result.rows.len(); @@ -6454,6 +6503,7 @@ where &pk_columns, request.target_catalog.as_deref(), overrides_postgres_system_values, + mysql_spatial_markers, )?; for (statement_index, batch_sql) in write_statements.iter().enumerate() { execute_transfer_write_statement( @@ -9866,6 +9916,7 @@ SELECT 1 FROM dual"# &[], None, false, + false, ) .unwrap(); @@ -9889,6 +9940,7 @@ SELECT 1 FROM dual"# &[], None, false, + false, ) .unwrap(); @@ -9995,6 +10047,7 @@ SELECT 1 FROM dual"# &[String::from("id")], None, false, + false, ) .unwrap(); @@ -10002,6 +10055,110 @@ SELECT 1 FROM dual"# assert!(statements[0].contains("ON DUPLICATE KEY UPDATE")); } + #[test] + fn mysql_spatial_transfer_reuses_validated_wkb_markers_for_all_modes() { + let columns = [String::from("id"), String::from("location"), String::from("name")]; + let column_types = [Some(String::from("int")), Some(String::from("point")), Some(String::from("varchar(32)"))]; + let rows = [vec![json!(1), json!("DBX_WKB:4326:0101000000000000000000F03F0000000000000040"), json!("alpha")]]; + + for mode in [TransferMode::Append, TransferMode::Overwrite, TransferMode::Upsert] { + let statements = generate_transfer_write_sql_batches( + &mode, + &columns, + &column_types, + &rows, + "places", + "", + &DatabaseType::Mysql, + &[String::from("id")], + None, + false, + true, + ) + .unwrap(); + + assert_eq!(statements.len(), 1); + assert!(statements[0].contains("ST_GeomFromWKB(0x0101000000000000000000F03F0000000000000040, 4326)")); + assert!(statements[0].contains("'alpha'")); + if mode == TransferMode::Upsert { + assert!(statements[0].contains("ON DUPLICATE KEY UPDATE")); + } + } + } + + #[test] + fn mysql_spatial_transfer_rejects_invalid_markers_and_keeps_public_insert_shape() { + let invalid = json!("DBX_WKB:4326:0101000000"); + let transfer = generate_transfer_write_sql_batches( + &TransferMode::Append, + &[String::from("location")], + &[Some(String::from("point"))], + &[vec![invalid.clone()]], + "places", + "", + &DatabaseType::Mysql, + &[], + None, + false, + true, + ) + .unwrap(); + let public = generate_insert_typed( + &[String::from("location")], + &[Some(String::from("point"))], + &[vec![invalid]], + "places", + "", + &DatabaseType::Mysql, + None, + ); + + assert_eq!(transfer, vec!["INSERT INTO `places` (`location`) VALUES\n('DBX_WKB:4326:0101000000')"]); + assert_eq!(public, transfer[0]); + assert!(!transfer[0].contains("ST_GeomFromWKB")); + } + + #[test] + fn mysql_spatial_transfer_projection_is_native_mysql_to_mysql_only() { + let sql = "SELECT `id`, `location` FROM `places` ORDER BY `id` LIMIT 2 OFFSET 0".to_string(); + let columns = [String::from("id"), String::from("location")]; + let column_types = [Some(String::from("int")), Some(String::from("point"))]; + + let (native_sql, native_markers) = mysql_spatial_transfer_select_sql( + sql.clone(), + &columns, + &column_types, + &DatabaseType::Mysql, + &DatabaseType::Mysql, + ); + assert!(native_markers); + assert!(native_sql.contains("CONCAT('DBX_WKB:', ST_SRID(`location`), ':', HEX(ST_AsWKB(`location`)))")); + assert!(native_sql.ends_with("ORDER BY `id` LIMIT 2 OFFSET 0")); + + let nonspatial_types = [Some(String::from("int")), Some(String::from("varchar(32)"))]; + assert_eq!( + mysql_spatial_transfer_select_sql( + sql.clone(), + &columns, + &nonspatial_types, + &DatabaseType::Mysql, + &DatabaseType::Mysql, + ), + (sql.clone(), false) + ); + + for (source, target) in [ + (DatabaseType::Mysql, DatabaseType::Postgres), + (DatabaseType::Postgres, DatabaseType::Mysql), + (DatabaseType::Doris, DatabaseType::Mysql), + ] { + assert_eq!( + mysql_spatial_transfer_select_sql(sql.clone(), &columns, &column_types, &source, &target), + (sql.clone(), false) + ); + } + } + #[test] fn postgres_transfer_insert_overrides_generated_always_identity_values() { for mode in [TransferMode::Append, TransferMode::Overwrite] { @@ -10016,6 +10173,7 @@ SELECT 1 FROM dual"# &[], None, true, + false, ) .unwrap(); @@ -10040,6 +10198,7 @@ SELECT 1 FROM dual"# &[String::from("id")], None, true, + false, ) .unwrap(); @@ -10063,6 +10222,7 @@ SELECT 1 FROM dual"# &[], None, false, + false, ) .unwrap(); @@ -10082,6 +10242,7 @@ SELECT 1 FROM dual"# &[], None, true, + false, ) .unwrap(); diff --git a/crates/dbx-core/tests/live_mysql_transfer.rs b/crates/dbx-core/tests/live_mysql_transfer.rs new file mode 100644 index 0000000000..59ef648fb5 --- /dev/null +++ b/crates/dbx-core/tests/live_mysql_transfer.rs @@ -0,0 +1,253 @@ +use dbx_core::connection::AppState; +use dbx_core::db::mysql; +use dbx_core::models::connection::{ConnectionConfig, DatabaseType}; +use dbx_core::storage::Storage; +use dbx_core::transfer::{ + transfer_table, TransferContent, TransferMode, TransferOwnershipPolicy, TransferRequest, TransferTableNameCase, +}; +use serde_json::json; +use std::time::Duration; + +fn live_mysql_config(id: &str) -> ConnectionConfig { + let host = std::env::var("DBX_LIVE_MYSQL_TRANSFER_HOST").expect("DBX_LIVE_MYSQL_TRANSFER_HOST"); + let port = + std::env::var("DBX_LIVE_MYSQL_TRANSFER_PORT").ok().and_then(|value| value.parse::().ok()).unwrap_or(3306); + let username = std::env::var("DBX_LIVE_MYSQL_TRANSFER_USER").unwrap_or_else(|_| "root".to_string()); + let password = std::env::var("DBX_LIVE_MYSQL_TRANSFER_PASSWORD").expect("DBX_LIVE_MYSQL_TRANSFER_PASSWORD"); + + serde_json::from_value(json!({ + "id": id, + "name": id, + "db_type": DatabaseType::Mysql, + "host": host, + "port": port, + "username": username, + "password": password, + "database": null, + "connect_timeout_secs": 10, + "query_timeout_secs": 30, + "idle_timeout_secs": 60, + "keepalive_interval_secs": 0 + })) + .expect("live MySQL transfer config should deserialize") +} + +fn mysql_url(config: &ConnectionConfig) -> String { + format!("mysql://{}:{}@{}:{}", config.username, config.password, config.host, config.port) +} + +fn transfer_request( + transfer_id: String, + connection_id: &str, + source_database: &str, + target_database: &str, + mode: TransferMode, +) -> TransferRequest { + TransferRequest { + transfer_id, + source_connection_id: connection_id.to_string(), + source_database: source_database.to_string(), + source_schema: source_database.to_string(), + source_catalog: None, + target_connection_id: connection_id.to_string(), + target_database: target_database.to_string(), + target_schema: target_database.to_string(), + target_catalog: None, + tables: vec!["spatial_matrix".to_string()], + create_table: false, + content: TransferContent::default(), + objects: Vec::new(), + mode, + target_table_name_case: TransferTableNameCase::Preserve, + ownership_policy: TransferOwnershipPolicy::Preserve, + batch_size: 2, + } +} + +async fn query_text(pool: &mysql::MySqlPool, sql: &str) -> String { + mysql::execute_query(pool, sql, false).await.unwrap().rows[0][0].as_str().unwrap().to_string() +} + +#[tokio::test] +#[ignore = "requires a disposable MySQL 5.7+ endpoint via DBX_LIVE_MYSQL_TRANSFER_* variables"] +async fn live_mysql_transfer_preserves_spatial_values_and_modes() { + let suffix = uuid::Uuid::new_v4().simple().to_string(); + let connection_id = format!("live-mysql-transfer-{suffix}"); + let source_database = format!("dbx_transfer_src_{}", &suffix[..12]); + let target_database = format!("dbx_transfer_dst_{}", &suffix[..12]); + let config = live_mysql_config(&connection_id); + let setup_pool = mysql::connect(&mysql_url(&config), Duration::from_secs(10)).await.unwrap(); + + let setup = format!( + "CREATE DATABASE `{source_database}`;\ + CREATE DATABASE `{target_database}`;\ + CREATE TABLE `{source_database}`.`spatial_matrix` (\ + id INT PRIMARY KEY, name VARCHAR(32), payload VARBINARY(8), created_at DATETIME,\ + g GEOMETRY NULL, p POINT NULL, ls LINESTRING NULL, poly POLYGON NULL,\ + mp MULTIPOINT NULL, mls MULTILINESTRING NULL, mpoly MULTIPOLYGON NULL,\ + gc GEOMETRYCOLLECTION NULL\ + );\ + CREATE TABLE `{target_database}`.`spatial_matrix` LIKE `{source_database}`.`spatial_matrix`;\ + INSERT INTO `{source_database}`.`spatial_matrix` VALUES (\ + 1, 'alpha', 0x00FF, '2026-08-12 10:20:30',\ + ST_GeomFromText('POINT(1 2)', 4326), ST_GeomFromText('POINT(3 4)', 4326),\ + ST_GeomFromText('LINESTRING(0 0,1 1)', 4326),\ + ST_GeomFromText('POLYGON((0 0,0 1,1 1,0 0))', 4326),\ + ST_GeomFromText('MULTIPOINT((0 0),(1 1))', 4326),\ + ST_GeomFromText('MULTILINESTRING((0 0,1 1))', 4326),\ + ST_GeomFromText('MULTIPOLYGON(((0 0,0 1,1 1,0 0)))', 4326),\ + ST_GeomFromText('GEOMETRYCOLLECTION(POINT(1 1))', 4326)\ + ), (2, 'nulls', NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL), (\ + 3, 'omega', 0xCAFE, '2026-08-12 11:22:33',\ + ST_GeomFromText('LINESTRING(2 2,3 3)', 3857), ST_GeomFromText('POINT(5 6)', 3857),\ + ST_GeomFromText('LINESTRING(4 4,5 5)', 3857),\ + ST_GeomFromText('POLYGON((2 2,2 3,3 3,2 2))', 3857),\ + ST_GeomFromText('MULTIPOINT((2 2),(3 3))', 3857),\ + ST_GeomFromText('MULTILINESTRING((2 2,3 3))', 3857),\ + ST_GeomFromText('MULTIPOLYGON(((2 2,2 3,3 3,2 2)))', 3857),\ + ST_GeomFromText('GEOMETRYCOLLECTION(POINT(2 2))', 3857)\ + )" + ); + mysql::execute_query(&setup_pool, &setup, true).await.unwrap(); + + let dir = std::env::temp_dir().join(format!("dbx-live-mysql-transfer-{suffix}")); + std::fs::create_dir_all(&dir).unwrap(); + let storage = Storage::open(&dir.join("storage.db")).await.unwrap(); + let state = AppState::new(storage); + state.configs.write().await.insert(connection_id.clone(), config); + let source_pool_key = state.get_or_create_pool(&connection_id, Some(&source_database)).await.unwrap(); + let target_pool_key = state.get_or_create_pool(&connection_id, Some(&target_database)).await.unwrap(); + + let test_result = async { + let append = transfer_table( + &state, + &transfer_request( + format!("live-mysql-transfer-append-{suffix}"), + &connection_id, + &source_database, + &target_database, + TransferMode::Append, + ), + "spatial_matrix", + 0, + &DatabaseType::Mysql, + &DatabaseType::Mysql, + &source_pool_key, + &target_pool_key, + |_| {}, + ) + .await?; + assert_eq!(append, 3); + assert_eq!( + query_text( + &setup_pool, + &format!( + "SELECT CAST(CONCAT((SELECT COUNT(*) FROM `{target_database}`.`spatial_matrix`), ':', \ + ST_AsText(p), ':', ST_SRID(p), ':', HEX(payload), ':', \ + DATE_FORMAT(created_at, '%Y-%m-%d %H:%i:%s'), ':', \ + (SELECT SUM(g IS NULL) FROM `{target_database}`.`spatial_matrix`)) AS CHAR) \ + FROM `{target_database}`.`spatial_matrix` WHERE id=1" + ), + ) + .await, + "3:POINT(3 4):4326:00FF:2026-08-12 10:20:30:1" + ); + let spatial_types = query_text( + &setup_pool, + &format!( + "SELECT CAST(CONCAT_WS('|', ST_AsText(g), ST_AsText(p), ST_AsText(ls), ST_AsText(poly), \ + ST_AsText(mp), ST_AsText(mls), ST_AsText(mpoly), ST_AsText(gc)) AS CHAR) \ + FROM `{target_database}`.`spatial_matrix` WHERE id=1" + ), + ) + .await; + assert_eq!( + spatial_types, + "POINT(1 2)|POINT(3 4)|LINESTRING(0 0,1 1)|POLYGON((0 0,0 1,1 1,0 0))|MULTIPOINT((0 0),(1 1))|MULTILINESTRING((0 0,1 1))|MULTIPOLYGON(((0 0,0 1,1 1,0 0)))|GEOMETRYCOLLECTION(POINT(1 1))" + ); + + mysql::execute_query( + &setup_pool, + &format!("INSERT INTO `{target_database}`.`spatial_matrix` (id, name) VALUES (99, 'stale')"), + false, + ) + .await?; + let overwrite = transfer_table( + &state, + &transfer_request( + format!("live-mysql-transfer-overwrite-{suffix}"), + &connection_id, + &source_database, + &target_database, + TransferMode::Overwrite, + ), + "spatial_matrix", + 0, + &DatabaseType::Mysql, + &DatabaseType::Mysql, + &source_pool_key, + &target_pool_key, + |_| {}, + ) + .await?; + assert_eq!(overwrite, 3); + assert_eq!( + query_text( + &setup_pool, + &format!("SELECT CAST(CONCAT(COUNT(*), ':', SUM(id=99)) AS CHAR) FROM `{target_database}`.`spatial_matrix`"), + ) + .await, + "3:0" + ); + + mysql::execute_query( + &setup_pool, + &format!("UPDATE `{source_database}`.`spatial_matrix` SET name='updated' WHERE id=1"), + false, + ) + .await?; + let upsert = transfer_table( + &state, + &transfer_request( + format!("live-mysql-transfer-upsert-{suffix}"), + &connection_id, + &source_database, + &target_database, + TransferMode::Upsert, + ), + "spatial_matrix", + 0, + &DatabaseType::Mysql, + &DatabaseType::Mysql, + &source_pool_key, + &target_pool_key, + |_| {}, + ) + .await?; + assert_eq!(upsert, 3); + assert_eq!( + query_text( + &setup_pool, + &format!( + "SELECT CAST(CONCAT(name, ':', ST_AsText(p), ':', ST_SRID(p)) AS CHAR) \ + FROM `{target_database}`.`spatial_matrix` WHERE id=1" + ), + ) + .await, + "updated:POINT(3 4):4326" + ); + Ok::<_, String>(()) + } + .await; + + let cleanup = mysql::execute_query( + &setup_pool, + &format!("DROP DATABASE `{source_database}`; DROP DATABASE `{target_database}`"), + true, + ) + .await; + setup_pool.disconnect().await.unwrap(); + let _ = std::fs::remove_dir_all(dir); + cleanup.unwrap(); + test_result.unwrap(); +}