fix(transfer): preserve MySQL spatial values

Closes #6004
This commit is contained in:
t8y2 committed 2026-08-12 11:00:07 +00:00
1 parent a9299e947f
commit 42afd3cad2
3 files changed
+424 -10

No files matched your search

+2 -2
View File
@@ -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:<srid>:<hex>` 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<DatabaseType>,
column_type: Option<&str>,
@@ -1487,7 +1487,7 @@ fn database_export_select_list(columns: &[String], column_types: &[Option<String
.join(", ")
}
fn replace_database_export_select_list(
pub(crate) fn replace_database_export_select_list(
sql: String,
columns: &[String],
column_types: &[Option<String>],
+169 -8
View File
@@ -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<serde_json::Value>],
column_types: &[Option<String>],
db_type: &DatabaseType,
mysql_spatial_markers: bool,
) -> Vec<String> {
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::<Vec<_>>().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<String>],
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<Vec<(String, usize)>, 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::<Vec<_>>();
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<Vec<String>, 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();
@@ -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::<u16>().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();
}