github/dbx
4
0 Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-11-03 00:32:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mysql): normalize temporal SQL literals

Browse Source
This commit is contained in:
t8y2 committed 2026-05-21 14:14:38 +08:00
1 parent eef6934e71
commit cb7682e932
7 files changed
+537 -47
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/dbx-core/src/database_export.rs
+6 -3
View File
@@ -187,7 +187,7 @@ pub async fn export_database_sql_core(
// Export data
if request.include_data {
// Get columns
let col_names: Vec<String> = match crate::schema::get_columns_core(
let columns = match crate::schema::get_columns_core(
state,
&request.connection_id,
&request.database,
@@ -196,7 +196,7 @@ pub async fn export_database_sql_core(
)
.await
{
Ok(cols) => cols.iter().map(|c| c.name.clone()).collect(),
Ok(cols) => cols,
Err(e) => {
writeln!(file, "-- ERROR exporting table {table_name}: {e}")
.map_err(|e| format!("Failed to write file: {e}"))?;
@@ -204,6 +204,8 @@ pub async fn export_database_sql_core(
continue;
}
};
let col_names = columns.iter().map(|c| c.name.clone()).collect::<Vec<_>>();
let col_types = columns.iter().map(|c| Some(c.data_type.clone())).collect::<Vec<_>>();
if !col_names.is_empty() {
// Get row count
@@ -260,8 +262,9 @@ pub async fn export_database_sql_core(
break;
}
let insert_sql = crate::transfer::generate_insert(
let insert_sql = crate::transfer::generate_insert_typed(
&col_names,
&col_types,
&result.rows,
table_name,
&request.schema,
crates/dbx-core/src/db/mysql.rs
+12 -1
View File
@@ -71,7 +71,7 @@ fn mysql_temporal_to_json_value(row: &MySqlRow, idx: usize) -> Option<serde_json
return Some(serde_json::Value::String(v.to_string()));
}
if let Ok(v) = row.try_get::<DateTime<Utc>, _>(idx) {
return Some(serde_json::Value::String(v.to_rfc3339()));
return Some(serde_json::Value::String(mysql_datetime_to_string(v)));
}
if let Ok(v) = row.try_get::<NaiveDate, _>(idx) {
return Some(serde_json::Value::String(v.to_string()));
@@ -82,6 +82,10 @@ fn mysql_temporal_to_json_value(row: &MySqlRow, idx: usize) -> Option<serde_json
None
}
fn mysql_datetime_to_string(value: DateTime<Utc>) -> String {
value.naive_utc().to_string()
}
fn mysql_value_to_json(row: &MySqlRow, idx: usize, type_name: &str) -> serde_json::Value {
if row.try_get_raw(idx).map(|v| v.is_null()).unwrap_or(true) {
return serde_json::Value::Null;
@@ -659,4 +663,11 @@ mod tests {
assert!(mysql_url_has_timezone_param("mysql://root@localhost/app?charset=utf8mb4&time-zone=%2B08:00"));
assert!(!mysql_url_has_timezone_param("mysql://root@localhost/app?charset=utf8mb4"));
}
#[test]
fn mysql_datetime_utc_values_display_without_rfc3339_offset() {
let value = DateTime::from_timestamp(1_778_544_000, 0).expect("valid timestamp");
assert_eq!(mysql_datetime_to_string(value), "2026-05-12 00:00:00");
}
}
crates/dbx-core/src/table_import.rs
+66 -3
View File
@@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize};
use crate::connection::AppState;
use crate::models::connection::DatabaseType;
use crate::transfer::{execute_on_pool, generate_insert, qualified_table};
use crate::transfer::{execute_on_pool, generate_insert_typed, get_columns_for_transfer, qualified_table};
pub const DEFAULT_PREVIEW_LIMIT: usize = 50;
pub const DEFAULT_BATCH_SIZE: usize = 500;
@@ -347,6 +347,7 @@ pub fn mapping_indexes(
pub fn build_import_insert_batches(
data: &ParsedImportFile,
mappings: &[TableImportColumnMapping],
target_column_types: &[(String, String)],
table: &str,
schema: &str,
db_type: &DatabaseType,
@@ -354,6 +355,15 @@ pub fn build_import_insert_batches(
) -> Result<Vec<ImportSqlBatch>, String> {
let mapped = mapping_indexes(data, mappings)?;
let columns = mapped.iter().map(|(_, target)| target.clone()).collect::<Vec<_>>();
let column_types = columns
.iter()
.map(|column| {
target_column_types
.iter()
.find(|(name, _)| name.eq_ignore_ascii_case(column))
.map(|(_, data_type)| data_type.clone())
})
.collect::<Vec<_>>();
let batch_size = batch_size.max(1);
let mut batches = Vec::new();
@@ -367,7 +377,7 @@ pub fn build_import_insert_batches(
.collect::<Vec<_>>()
})
.collect::<Vec<_>>();
let sql = generate_insert(&columns, &rows, table, schema, db_type);
let sql = generate_insert_typed(&columns, &column_types, &rows, table, schema, db_type);
if !sql.trim().is_empty() {
batches.push(ImportSqlBatch { sql, row_count: chunk.len() });
}
@@ -439,9 +449,24 @@ where
error: None,
});
let target_column_types = get_columns_for_transfer(
state,
pool_key,
&request.connection_id,
&request.database,
&request.schema,
&request.table,
)
.await
.unwrap_or_default()
.into_iter()
.map(|column| (column.name, column.data_type))
.collect::<Vec<_>>();
let batches = match build_import_insert_batches(
&parsed,
&request.mappings,
&target_column_types,
&request.table,
&request.schema,
db_type,
@@ -586,7 +611,7 @@ mod tests {
};
let batches =
build_import_insert_batches(&data, &mappings, "users", "public", &DatabaseType::Postgres, 2).unwrap();
build_import_insert_batches(&data, &mappings, &[], "users", "public", &DatabaseType::Postgres, 2).unwrap();
assert_eq!(batches, vec![
ImportSqlBatch {
@@ -599,4 +624,42 @@ mod tests {
},
]);
}
#[test]
fn import_insert_batches_use_target_column_types_for_mysql_temporal_values() {
let mappings = vec![
TableImportColumnMapping {
source_column: "start".to_string(),
target_column: "insurance_start_time".to_string(),
},
TableImportColumnMapping { source_column: "raw".to_string(), target_column: "raw_text".to_string() },
];
let data = ParsedImportFile {
columns: vec!["start".to_string(), "raw".to_string()],
rows: vec![vec![
serde_json::json!("2026-05-12T00:00:00+00:00"),
serde_json::json!("2026-05-12T00:00:00+00:00"),
]],
total_rows: 1,
};
let batches = build_import_insert_batches(
&data,
&mappings,
&[
("insurance_start_time".to_string(), "datetime".to_string()),
("raw_text".to_string(), "varchar(64)".to_string()),
],
"policies",
"",
&DatabaseType::Mysql,
500,
)
.unwrap();
assert_eq!(batches, vec![ImportSqlBatch {
sql: "INSERT INTO `policies` (`insurance_start_time`, `raw_text`) VALUES\n('2026-05-12 00:00:00', '2026-05-12T00:00:00+00:00')".to_string(),
row_count: 1,
}]);
}
}
crates/dbx-core/src/transfer.rs
+248 -21
View File
@@ -78,6 +78,10 @@ pub fn qualified_table(table: &str, schema: &str, db_type: &DatabaseType) -> Str
}
pub fn escape_value(val: &serde_json::Value, db_type: &DatabaseType) -> String {
escape_value_typed(val, db_type, None)
}
pub fn escape_value_typed(val: &serde_json::Value, db_type: &DatabaseType, column_type: Option<&str>) -> String {
match val {
serde_json::Value::Null => "NULL".to_string(),
serde_json::Value::Bool(b) => match db_type {
@@ -102,7 +106,7 @@ pub fn escape_value(val: &serde_json::Value, db_type: &DatabaseType) -> String {
},
serde_json::Value::Number(n) => n.to_string(),
serde_json::Value::String(s) => {
format!("'{}'", s.replace('\'', "''"))
format!("'{}'", format_literal_string(s, db_type, column_type).replace('\'', "''"))
}
_ => {
let s = val.to_string();
@@ -111,6 +115,127 @@ pub fn escape_value(val: &serde_json::Value, db_type: &DatabaseType) -> String {
}
}
fn format_literal_string(value: &str, db_type: &DatabaseType, column_type: Option<&str>) -> String {
if is_mysql_datetime_literal_database(db_type) && column_type.map(is_temporal_column_type).unwrap_or(true) {
normalize_mysql_temporal_literal(value, column_type).unwrap_or_else(|| value.to_string())
} else {
value.to_string()
}
}
fn is_mysql_datetime_literal_database(db_type: &DatabaseType) -> bool {
matches!(
db_type,
DatabaseType::Mysql
| DatabaseType::Doris
| DatabaseType::StarRocks
| DatabaseType::Goldendb
| DatabaseType::Sundb
)
}
fn normalize_mysql_temporal_literal(value: &str, column_type: Option<&str>) -> Option<String> {
let bytes = value.as_bytes();
if bytes.len() < 20 || !is_mysql_datetime_base(bytes) {
return None;
}
let rest = &value[19..];
let (fraction, offset) = if let Some(after_dot) = rest.strip_prefix('.') {
let digit_count = after_dot.bytes().take_while(|b| b.is_ascii_digit()).count();
if digit_count == 0 {
return None;
}
let fraction_len = 1 + digit_count;
(&rest[..fraction_len.min(7)], &rest[fraction_len..])
} else {
("", rest)
};
if !is_timezone_suffix(offset) {
return None;
}
match temporal_column_kind(column_type) {
Some("date") => Some(value[..10].to_string()),
Some("time") => Some(format!("{}{}", &value[11..19], fraction)),
_ => Some(format!("{} {}{}", &value[..10], &value[11..19], fraction)),
}
}
fn is_temporal_column_type(column_type: &str) -> bool {
temporal_column_kind(Some(column_type)).is_some()
}
fn temporal_column_kind(column_type: Option<&str>) -> Option<&'static str> {
let base = column_type?.trim().to_ascii_lowercase();
let base = base.split(['(', ':', ' ']).next().unwrap_or("");
match base {
"date" => Some("date"),
"time" => Some("time"),
"datetime" | "timestamp" => Some("datetime"),
_ => None,
}
}
fn is_mysql_datetime_base(bytes: &[u8]) -> bool {
matches!(
bytes,
[
y0,
y1,
y2,
y3,
b'-',
m0,
m1,
b'-',
d0,
d1,
sep,
h0,
h1,
b':',
min0,
min1,
b':',
s0,
s1,
..
] if y0.is_ascii_digit()
&& y1.is_ascii_digit()
&& y2.is_ascii_digit()
&& y3.is_ascii_digit()
&& m0.is_ascii_digit()
&& m1.is_ascii_digit()
&& d0.is_ascii_digit()
&& d1.is_ascii_digit()
&& (*sep == b'T' || *sep == b' ')
&& h0.is_ascii_digit()
&& h1.is_ascii_digit()
&& min0.is_ascii_digit()
&& min1.is_ascii_digit()
&& s0.is_ascii_digit()
&& s1.is_ascii_digit()
)
}
fn is_timezone_suffix(value: &str) -> bool {
if value.eq_ignore_ascii_case("z") {
return true;
}
let bytes = value.as_bytes();
matches!(
bytes,
[sign, h0, h1, b':', m0, m1]
if (*sign == b'+' || *sign == b'-')
&& h0.is_ascii_digit()
&& h1.is_ascii_digit()
&& m0.is_ascii_digit()
&& m1.is_ascii_digit()
)
}
pub fn map_column_type(source_type: &str, _source_db: &DatabaseType, target_db: &DatabaseType) -> String {
let t = source_type.to_lowercase();
let base = t.split('(').next().unwrap_or(&t).trim();
@@ -277,6 +402,17 @@ pub fn generate_insert(
table: &str,
schema: &str,
db_type: &DatabaseType,
) -> String {
generate_insert_typed(columns, &vec![None; columns.len()], rows, table, schema, db_type)
}
pub fn generate_insert_typed(
columns: &[String],
column_types: &[Option<String>],
rows: &[Vec<serde_json::Value>],
table: &str,
schema: &str,
db_type: &DatabaseType,
) -> String {
if rows.is_empty() {
return String::new();
@@ -285,17 +421,30 @@ pub fn generate_insert(
let full_table = qualified_table(table, schema, db_type);
let col_list = columns.iter().map(|c| quote_identifier(c, db_type)).collect::<Vec<_>>().join(", ");
let value_rows: Vec<String> = rows
.iter()
.map(|row| {
let vals: Vec<String> = row.iter().map(|v| escape_value(v, db_type)).collect();
format!("({})", vals.join(", "))
})
.collect();
let value_rows = value_rows_sql(rows, column_types, db_type);
format!("INSERT INTO {full_table} ({col_list}) VALUES\n{}", value_rows.join(",\n"))
}
fn value_rows_sql(
rows: &[Vec<serde_json::Value>],
column_types: &[Option<String>],
db_type: &DatabaseType,
) -> Vec<String> {
rows.iter()
.map(|row| {
let vals: Vec<String> = row
.iter()
.enumerate()
.map(|(index, v)| {
escape_value_typed(v, db_type, column_types.get(index).and_then(|value| value.as_deref()))
})
.collect();
format!("({})", vals.join(", "))
})
.collect()
}
pub fn generate_upsert(
columns: &[String],
rows: &[Vec<serde_json::Value>],
@@ -303,6 +452,18 @@ pub fn generate_upsert(
schema: &str,
db_type: &DatabaseType,
pk_columns: &[String],
) -> String {
generate_upsert_typed(columns, &vec![None; columns.len()], rows, table, schema, db_type, pk_columns)
}
pub fn generate_upsert_typed(
columns: &[String],
column_types: &[Option<String>],
rows: &[Vec<serde_json::Value>],
table: &str,
schema: &str,
db_type: &DatabaseType,
pk_columns: &[String],
) -> String {
if rows.is_empty() || pk_columns.is_empty() {
return String::new();
@@ -311,13 +472,7 @@ pub fn generate_upsert(
let full_table = qualified_table(table, schema, db_type);
let col_list = columns.iter().map(|c| quote_identifier(c, db_type)).collect::<Vec<_>>().join(", ");
let value_rows: Vec<String> = rows
.iter()
.map(|row| {
let vals: Vec<String> = row.iter().map(|v| escape_value(v, db_type)).collect();
format!("({})", vals.join(", "))
})
.collect();
let value_rows = value_rows_sql(rows, column_types, db_type);
let non_pk_columns: Vec<&String> = columns.iter().filter(|c| !pk_columns.contains(c)).collect();
@@ -400,7 +555,18 @@ pub fn generate_upsert(
let vals: Vec<String> = row
.iter()
.zip(columns.iter())
.map(|(v, c)| format!("{} AS {}", escape_value(v, db_type), quote_identifier(c, db_type)))
.enumerate()
.map(|(index, (v, c))| {
format!(
"{} AS {}",
escape_value_typed(
v,
db_type,
column_types.get(index).and_then(|value| value.as_deref())
),
quote_identifier(c, db_type)
)
})
.collect();
format!("SELECT {} FROM dual", vals.join(", "))
})
@@ -436,7 +602,7 @@ pub fn generate_upsert(
sql.push_str(&format!("\nWHEN NOT MATCHED THEN INSERT ({insert_cols}) VALUES ({insert_vals})"));
sql
}
_ => generate_insert(columns, rows, table, schema, db_type),
_ => generate_insert_typed(columns, column_types, rows, table, schema, db_type),
}
}
@@ -712,6 +878,7 @@ where
}
let col_names: Vec<String> = columns.iter().map(|c| c.name.clone()).collect();
let col_types: Vec<Option<String>> = columns.iter().map(|c| Some(c.data_type.clone())).collect();
log::info!("[transfer] {} has {} columns, counting rows...", table, columns.len());
// Count source rows
@@ -800,10 +967,23 @@ where
}
let batch_sql = match effective_mode {
TransferMode::Upsert => {
generate_upsert(&col_names, &result.rows, table, &request.target_schema, target_db_type, &pk_columns)
}
_ => generate_insert(&col_names, &result.rows, table, &request.target_schema, target_db_type),
TransferMode::Upsert => generate_upsert_typed(
&col_names,
&col_types,
&result.rows,
table,
&request.target_schema,
target_db_type,
&pk_columns,
),
_ => generate_insert_typed(
&col_names,
&col_types,
&result.rows,
table,
&request.target_schema,
target_db_type,
),
};
if !batch_sql.is_empty() {
execute_on_pool(state, target_pool_key, &batch_sql)
@@ -833,3 +1013,50 @@ where
Ok(total_transferred)
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn mysql_insert_normalizes_rfc3339_datetime_strings() {
let sql = generate_insert_typed(
&[String::from("insurance_start_time")],
&[Some(String::from("datetime"))],
&[vec![json!("2026-05-12T00:00:00+00:00")]],
"policies",
"",
&DatabaseType::Mysql,
);
assert_eq!(sql, "INSERT INTO `policies` (`insurance_start_time`) VALUES\n('2026-05-12 00:00:00')");
}
#[test]
fn mysql_insert_uses_column_types_for_temporal_literals() {
let sql = generate_insert_typed(
&[String::from("dt"), String::from("raw_text"), String::from("d"), String::from("t")],
&[
Some(String::from("datetime")),
Some(String::from("varchar(64)")),
Some(String::from("date")),
Some(String::from("time")),
],
&[vec![
json!("2026-05-12T00:00:00+00:00"),
json!("2026-05-12T00:00:00+00:00"),
json!("2026-05-12T00:00:00+00:00"),
json!("2026-05-12T09:30:45+00:00"),
]],
"policies",
"",
&DatabaseType::Mysql,
);
assert_eq!(
sql,
"INSERT INTO `policies` (`dt`, `raw_text`, `d`, `t`) VALUES\n('2026-05-12 00:00:00', '2026-05-12T00:00:00+00:00', '2026-05-12', '09:30:45')"
);
}
}