github/dbx
4
0 Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-11-15 22:53:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(transfer): allow unquoted gaussdb transfer identifiers

Browse Source
This commit is contained in:
zipg authored and GitHub committed 2026-08-31 22:28:40 +08:00
1 parent 0f33449bdf
commit 4baa46ca88
22 files changed
+423 -53
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/dbx-core/src/sql_dialect.rs
+3 -1
View File
@@ -37,7 +37,9 @@ pub use descriptor::{
pub use identifiers::{
normalize_where_input, qualified_table_name, qualified_table_name_with_catalog, quote_table_identifier,
};
pub(crate) use identifiers::{parse_sqlserver_linked_schema_ref, qualified_transfer_table, quote_transfer_identifier};
pub(crate) use identifiers::{
parse_sqlserver_linked_schema_ref, qualified_transfer_table, quote_transfer_identifier, transfer_column_identifier,
};
pub use table_select::{
build_count_table_sql, build_table_data_select_sql, build_table_data_select_sql_with_database,
build_table_select_sql, DBX_LARGE_VALUE_BYTES_COLUMN_PREFIX,
crates/dbx-core/src/sql_dialect/identifiers.rs
+25
View File
@@ -395,6 +395,31 @@ pub(crate) fn quote_transfer_identifier(name: &str, database_type: &DatabaseType
}
}
pub(crate) fn transfer_column_identifier(
name: &str,
database_type: &DatabaseType,
quote_target_column_names: bool,
) -> String {
if quote_target_column_names
|| !matches!(database_type, DatabaseType::Gaussdb | DatabaseType::OpenGauss)
|| !is_simple_unquoted_identifier(name)
|| is_postgres_reserved_identifier(&name.to_ascii_lowercase())
{
quote_transfer_identifier(name, database_type)
} else {
name.to_string()
}
}
fn is_simple_unquoted_identifier(name: &str) -> bool {
let mut chars = name.chars();
let Some(first) = chars.next() else {
return false;
};
(first == '_' || first.is_ascii_alphabetic())
&& chars.all(|ch| ch == '_' || ch == '$' || ch.is_ascii_alphanumeric())
}
/// Qualified table name for transfer SQL.
///
/// * Without catalog: produces `schema.table` (or just `table` for MySQL family).
crates/dbx-core/src/transfer.rs
+341 -52
View File
@@ -13,7 +13,9 @@ use crate::query::{
agent_execute_query_params, pool_error_action, PoolErrorAction, QueryExecutionOptions, AGENT_PROTOCOL_MAX_ROWS,
};
use crate::sql::{split_sql_statements, split_sql_statements_for_database};
use crate::sql_dialect::{normalize_len_params, qualified_transfer_table, quote_transfer_identifier};
use crate::sql_dialect::{
normalize_len_params, qualified_transfer_table, quote_transfer_identifier, transfer_column_identifier,
};
static CANCELLED: std::sync::LazyLock<RwLock<HashSet<String>>> =
std::sync::LazyLock::new(|| RwLock::new(HashSet::new()));
@@ -185,11 +187,17 @@ pub struct TransferRequest {
pub mode: TransferMode,
#[serde(default)]
pub target_table_name_case: TransferTableNameCase,
#[serde(default = "default_quote_target_column_names")]
pub quote_target_column_names: bool,
#[serde(default)]
pub ownership_policy: TransferOwnershipPolicy,
pub batch_size: usize,
}
fn default_quote_target_column_names() -> bool {
true
}
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct TransferOwnershipPreview {
@@ -1491,7 +1499,12 @@ fn writable_transfer_columns(
.collect()
}
fn transfer_column_names_match(target_db_type: &DatabaseType, left: &str, right: &str) -> bool {
fn transfer_column_names_match(
target_db_type: &DatabaseType,
quote_target_column_names: bool,
left: &str,
right: &str,
) -> bool {
if matches!(
target_db_type,
DatabaseType::Mysql
@@ -1508,7 +1521,11 @@ fn transfer_column_names_match(target_db_type: &DatabaseType, left: &str, right:
| DatabaseType::Impala
| DatabaseType::Spark
| DatabaseType::Access
) {
) || (matches!(target_db_type, DatabaseType::Gaussdb | DatabaseType::OpenGauss) && !quote_target_column_names)
{
// Unquoted GaussDB/openGauss identifiers fold to lowercase server-side,
// so with quoting disabled a table created from mixed-case source
// columns reports folded names back from the catalog.
left.eq_ignore_ascii_case(right)
} else {
left == right
@@ -1519,11 +1536,14 @@ fn missing_transfer_target_columns(
target_columns: &[db::ColumnInfo],
col_names: &[String],
target_db_type: &DatabaseType,
quote_target_column_names: bool,
) -> Vec<String> {
col_names
.iter()
.filter(|name| {
!target_columns.iter().any(|column| transfer_column_names_match(target_db_type, name, &column.name))
!target_columns.iter().any(|column| {
transfer_column_names_match(target_db_type, quote_target_column_names, name, &column.name)
})
})
.cloned()
.collect()
@@ -1545,12 +1565,15 @@ fn required_unmapped_transfer_target_columns(
target_columns: &[db::ColumnInfo],
col_names: &[String],
target_db_type: &DatabaseType,
quote_target_column_names: bool,
) -> Vec<String> {
target_columns
.iter()
.filter(|column| {
!target_column_can_be_omitted(column, target_db_type)
&& !col_names.iter().any(|name| transfer_column_names_match(target_db_type, name, &column.name))
&& !col_names.iter().any(|name| {
transfer_column_names_match(target_db_type, quote_target_column_names, name, &column.name)
})
})
.map(|column| column.name.clone())
.collect()
@@ -3119,6 +3142,31 @@ pub fn generate_create_table_ddl(
source_db: &DatabaseType,
table_comment: Option<&str>,
catalog: Option<&str>,
) -> String {
generate_create_table_ddl_with_column_quoting(
columns,
table,
source_schema,
schema,
target_db,
source_db,
table_comment,
catalog,
true,
)
}
#[allow(clippy::too_many_arguments)]
fn generate_create_table_ddl_with_column_quoting(
columns: &[db::ColumnInfo],
table: &str,
source_schema: &str,
schema: &str,
target_db: &DatabaseType,
source_db: &DatabaseType,
table_comment: Option<&str>,
catalog: Option<&str>,
quote_target_column_names: bool,
) -> String {
let full_table = qualified_table(table, schema, target_db, catalog);
@@ -3135,7 +3183,11 @@ pub fn generate_create_table_ddl(
for c in columns {
col_lines.push({
let mapped_type = postgres_column_type_sql(c, source_schema, schema, source_db, target_db);
let mut line = format!(" {} {}", quote_identifier(&c.name, target_db), mapped_type);
let mut line = format!(
" {} {}",
transfer_column_identifier(&c.name, target_db, quote_target_column_names),
mapped_type
);
if let Some(default_clause) = column_default_clause(c, source_schema, schema, source_db, target_db) {
line.push(' ');
line.push_str(&default_clause);
@@ -3167,7 +3219,7 @@ pub fn generate_create_table_ddl(
if !matches!(target_db, DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala) {
for c in columns {
if c.is_primary_key {
let qname = quote_identifier(&c.name, target_db);
let qname = transfer_column_identifier(&c.name, target_db, quote_target_column_names);
if is_mysql_family {
let mapped = map_column_type(&c.data_type, source_db, target_db);
if mysql_type_needs_key_prefix(&mapped) {
@@ -3232,6 +3284,17 @@ pub fn generate_comment_ddl(
schema: &str,
target_db: &DatabaseType,
table_comment: Option<&str>,
) -> Vec<String> {
generate_comment_ddl_with_column_quoting(columns, table, schema, target_db, table_comment, true)
}
fn generate_comment_ddl_with_column_quoting(
columns: &[db::ColumnInfo],
table: &str,
schema: &str,
target_db: &DatabaseType,
table_comment: Option<&str>,
quote_target_column_names: bool,
) -> Vec<String> {
if !(is_postgres_transfer_dialect(target_db)
|| matches!(target_db, DatabaseType::Oracle | DatabaseType::ClickHouse))
@@ -3260,7 +3323,7 @@ pub fn generate_comment_ddl(
continue;
}
let escaped = trimmed.replace('\'', "''");
let qcol = quote_identifier(&c.name, target_db);
let qcol = transfer_column_identifier(&c.name, target_db, quote_target_column_names);
match target_db {
target_db if is_postgres_transfer_dialect(target_db) || matches!(target_db, DatabaseType::Oracle) => {
@@ -3329,9 +3392,26 @@ impl InsertSqlTemplate {
db_type: &DatabaseType,
catalog: Option<&str>,
overrides_postgres_system_values: bool,
) -> Self {
Self::new_with_column_quoting(columns, table, schema, db_type, catalog, overrides_postgres_system_values, true)
}
#[allow(clippy::too_many_arguments)]
fn new_with_column_quoting(
columns: &[String],
table: &str,
schema: &str,
db_type: &DatabaseType,
catalog: Option<&str>,
overrides_postgres_system_values: bool,
quote_target_column_names: bool,
) -> Self {
let full_table = qualified_table(table, schema, db_type, catalog);
let col_list = columns.iter().map(|column| quote_identifier(column, db_type)).collect::<Vec<_>>().join(", ");
let col_list = columns
.iter()
.map(|column| transfer_column_identifier(column, db_type, quote_target_column_names))
.collect::<Vec<_>>()
.join(", ");
let overriding = if overrides_postgres_system_values && matches!(db_type, DatabaseType::Postgres) {
" OVERRIDING SYSTEM VALUE"
} else {
@@ -3465,6 +3545,7 @@ pub fn generate_upsert_typed(
catalog,
false,
false,
true,
)
}
@@ -3480,13 +3561,18 @@ fn generate_upsert_typed_for_transfer(
catalog: Option<&str>,
overrides_postgres_system_values: bool,
mysql_spatial_markers: bool,
quote_target_column_names: bool,
) -> String {
if rows.is_empty() || pk_columns.is_empty() {
return String::new();
}
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 col_list = columns
.iter()
.map(|column| transfer_column_identifier(column, db_type, quote_target_column_names))
.collect::<Vec<_>>()
.join(", ");
let value_rows = value_rows_sql(rows, column_types, db_type, mysql_spatial_markers);
@@ -3502,7 +3588,11 @@ fn generate_upsert_typed_for_transfer(
if is_postgres_transfer_dialect(db_type)
|| matches!(db_type, DatabaseType::Sqlite | DatabaseType::CloudflareD1 | DatabaseType::DuckDb) =>
{
let pk_list = pk_columns.iter().map(|c| quote_identifier(c, db_type)).collect::<Vec<_>>().join(", ");
let pk_list = pk_columns
.iter()
.map(|column| transfer_column_identifier(column, db_type, quote_target_column_names))
.collect::<Vec<_>>()
.join(", ");
let overriding = if overrides_postgres_system_values && matches!(db_type, DatabaseType::Postgres) {
" OVERRIDING SYSTEM VALUE"
} else {
@@ -3516,7 +3606,7 @@ fn generate_upsert_typed_for_transfer(
let update_set = non_pk_columns
.iter()
.map(|c| {
let qc = quote_identifier(c, db_type);
let qc = transfer_column_identifier(c, db_type, quote_target_column_names);
format!("{qc} = EXCLUDED.{qc}")
})
.collect::<Vec<_>>()
@@ -3529,13 +3619,13 @@ fn generate_upsert_typed_for_transfer(
let mut sql = format!("INSERT INTO {full_table} ({col_list}) VALUES\n{}", value_rows.join(",\n"));
if non_pk_columns.is_empty() {
sql.push_str("\nON DUPLICATE KEY UPDATE ");
let first_pk = quote_identifier(&pk_columns[0], db_type);
let first_pk = transfer_column_identifier(&pk_columns[0], db_type, quote_target_column_names);
sql.push_str(&format!("{first_pk} = {first_pk}"));
} else {
let update_set = non_pk_columns
.iter()
.map(|c| {
let qc = quote_identifier(c, db_type);
let qc = transfer_column_identifier(c, db_type, quote_target_column_names);
format!("{qc} = VALUES({qc})")
})
.collect::<Vec<_>>()
@@ -3545,11 +3635,15 @@ fn generate_upsert_typed_for_transfer(
sql
}
DatabaseType::SqlServer => {
let src_col_list = columns.iter().map(|c| quote_identifier(c, db_type)).collect::<Vec<_>>().join(", ");
let src_col_list = columns
.iter()
.map(|column| transfer_column_identifier(column, db_type, quote_target_column_names))
.collect::<Vec<_>>()
.join(", ");
let on_clause = pk_columns
.iter()
.map(|c| {
let qc = quote_identifier(c, db_type);
let qc = transfer_column_identifier(c, db_type, quote_target_column_names);
format!("target.{qc} = src.{qc}")
})
.collect::<Vec<_>>()
@@ -3564,7 +3658,7 @@ fn generate_upsert_typed_for_transfer(
let update_set = non_pk_columns
.iter()
.map(|c| {
let qc = quote_identifier(c, db_type);
let qc = transfer_column_identifier(c, db_type, quote_target_column_names);
format!("target.{qc} = src.{qc}")
})
.collect::<Vec<_>>()
@@ -3572,9 +3666,16 @@ fn generate_upsert_typed_for_transfer(
sql.push_str(&format!("\nWHEN MATCHED THEN UPDATE SET {update_set}"));
}
let insert_cols = columns.iter().map(|c| quote_identifier(c, db_type)).collect::<Vec<_>>().join(", ");
let insert_vals =
columns.iter().map(|c| format!("src.{}", quote_identifier(c, db_type))).collect::<Vec<_>>().join(", ");
let insert_cols = columns
.iter()
.map(|column| transfer_column_identifier(column, db_type, quote_target_column_names))
.collect::<Vec<_>>()
.join(", ");
let insert_vals = columns
.iter()
.map(|column| format!("src.{}", transfer_column_identifier(column, db_type, quote_target_column_names)))
.collect::<Vec<_>>()
.join(", ");
sql.push_str(&format!("\nWHEN NOT MATCHED THEN INSERT ({insert_cols}) VALUES ({insert_vals});"));
sql
}
@@ -3586,7 +3687,7 @@ fn generate_upsert_typed_for_transfer(
vals.push(format!(
"{} AS {}",
escape_value_typed(v, db_type, column_types.get(index).and_then(|value| value.as_deref())),
quote_identifier(c, db_type)
transfer_column_identifier(c, db_type, quote_target_column_names)
));
}
using_rows.push(format!("SELECT {} FROM dual", vals.join(", ")));
@@ -3595,7 +3696,7 @@ fn generate_upsert_typed_for_transfer(
let on_clause = pk_columns
.iter()
.map(|c| {
let qc = quote_identifier(c, db_type);
let qc = transfer_column_identifier(c, db_type, quote_target_column_names);
format!("t.{qc} = s.{qc}")
})
.collect::<Vec<_>>()
@@ -3608,7 +3709,7 @@ fn generate_upsert_typed_for_transfer(
let update_set = non_pk_columns
.iter()
.map(|c| {
let qc = quote_identifier(c, db_type);
let qc = transfer_column_identifier(c, db_type, quote_target_column_names);
format!("t.{qc} = s.{qc}")
})
.collect::<Vec<_>>()
@@ -3616,14 +3717,29 @@ fn generate_upsert_typed_for_transfer(
sql.push_str(&format!("\nWHEN MATCHED THEN UPDATE SET {update_set}"));
}
let insert_cols = columns.iter().map(|c| quote_identifier(c, db_type)).collect::<Vec<_>>().join(", ");
let insert_vals =
columns.iter().map(|c| format!("s.{}", quote_identifier(c, db_type))).collect::<Vec<_>>().join(", ");
let insert_cols = columns
.iter()
.map(|column| transfer_column_identifier(column, db_type, quote_target_column_names))
.collect::<Vec<_>>()
.join(", ");
let insert_vals = columns
.iter()
.map(|column| format!("s.{}", transfer_column_identifier(column, db_type, quote_target_column_names)))
.collect::<Vec<_>>()
.join(", ");
sql.push_str(&format!("\nWHEN NOT MATCHED THEN INSERT ({insert_cols}) VALUES ({insert_vals})"));
sql
}
_ => {
let template = InsertSqlTemplate::new(columns, table, schema, db_type, catalog, false);
let template = InsertSqlTemplate::new_with_column_quoting(
columns,
table,
schema,
db_type,
catalog,
false,
quote_target_column_names,
);
template.build(&value_rows_sql(rows, column_types, db_type, mysql_spatial_markers))
}
}
@@ -3813,6 +3929,7 @@ fn generate_transfer_write_sql(
catalog: Option<&str>,
overrides_postgres_system_values: bool,
mysql_spatial_markers: bool,
quote_target_column_names: bool,
) -> String {
match mode {
TransferMode::Upsert => generate_upsert_typed_for_transfer(
@@ -3826,13 +3943,21 @@ fn generate_transfer_write_sql(
catalog,
overrides_postgres_system_values,
mysql_spatial_markers,
quote_target_column_names,
),
_ => {
if rows.is_empty() {
return String::new();
}
let template =
InsertSqlTemplate::new(columns, table, schema, db_type, catalog, overrides_postgres_system_values);
let template = InsertSqlTemplate::new_with_column_quoting(
columns,
table,
schema,
db_type,
catalog,
overrides_postgres_system_values,
quote_target_column_names,
);
template.build(&value_rows_sql(rows, column_types, db_type, mysql_spatial_markers))
}
}
@@ -3864,7 +3989,9 @@ pub(crate) fn generate_insert_typed_sql_batches_from_value_rows(
catalog: Option<&str>,
limits: SqlBatchLimits,
) -> Result<Vec<(String, usize)>, String> {
generate_insert_sql_batches_from_value_rows(columns, value_rows, table, schema, db_type, catalog, limits, false)
generate_insert_sql_batches_from_value_rows(
columns, value_rows, table, schema, db_type, catalog, limits, false, true,
)
}
#[allow(clippy::too_many_arguments)]
@@ -3879,6 +4006,7 @@ fn generate_insert_typed_sql_batches_for_transfer(
limits: SqlBatchLimits,
overrides_postgres_system_values: bool,
mysql_spatial_markers: bool,
quote_target_column_names: bool,
) -> Result<Vec<(String, usize)>, String> {
if rows.is_empty() {
return Ok(Vec::new());
@@ -3894,6 +4022,7 @@ fn generate_insert_typed_sql_batches_for_transfer(
catalog,
limits,
overrides_postgres_system_values,
quote_target_column_names,
)
}
@@ -3907,6 +4036,7 @@ fn generate_insert_sql_batches_from_value_rows(
catalog: Option<&str>,
limits: SqlBatchLimits,
overrides_postgres_system_values: bool,
quote_target_column_names: bool,
) -> Result<Vec<(String, usize)>, String> {
if value_rows.is_empty() {
return Ok(Vec::new());
@@ -3919,7 +4049,15 @@ fn generate_insert_sql_batches_from_value_rows(
});
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 template = InsertSqlTemplate::new_with_column_quoting(
columns,
table,
schema,
db_type,
catalog,
overrides_postgres_system_values,
quote_target_column_names,
);
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;
@@ -3957,6 +4095,7 @@ fn generate_insert_sql_batches_from_value_rows(
}
#[allow(clippy::too_many_arguments)]
#[cfg_attr(not(test), allow(dead_code))]
fn generate_transfer_write_sql_batches(
mode: &TransferMode,
columns: &[String],
@@ -3969,6 +4108,37 @@ fn generate_transfer_write_sql_batches(
catalog: Option<&str>,
overrides_postgres_system_values: bool,
mysql_spatial_markers: bool,
) -> Result<Vec<String>, String> {
generate_transfer_write_sql_batches_with_column_quoting(
mode,
columns,
column_types,
rows,
table,
schema,
db_type,
pk_columns,
catalog,
overrides_postgres_system_values,
mysql_spatial_markers,
true,
)
}
#[allow(clippy::too_many_arguments)]
fn generate_transfer_write_sql_batches_with_column_quoting(
mode: &TransferMode,
columns: &[String],
column_types: &[Option<String>],
rows: &[Vec<serde_json::Value>],
table: &str,
schema: &str,
db_type: &DatabaseType,
pk_columns: &[String],
catalog: Option<&str>,
overrides_postgres_system_values: bool,
mysql_spatial_markers: bool,
quote_target_column_names: bool,
) -> Result<Vec<String>, String> {
if rows.is_empty() {
return Ok(Vec::new());
@@ -3986,6 +4156,7 @@ fn generate_transfer_write_sql_batches(
SqlBatchLimits::for_database(db_type, max_transfer_write_rows(db_type, mode)),
overrides_postgres_system_values,
mysql_spatial_markers,
quote_target_column_names,
)?
.into_iter()
.map(|(sql, _)| sql)
@@ -4014,6 +4185,7 @@ fn generate_transfer_write_sql_batches(
catalog,
overrides_postgres_system_values,
mysql_spatial_markers,
quote_target_column_names,
);
while end < rows.len() && end - start < max_rows {
@@ -4029,6 +4201,7 @@ fn generate_transfer_write_sql_batches(
catalog,
overrides_postgres_system_values,
mysql_spatial_markers,
quote_target_column_names,
);
if candidate.len() > max_sql_bytes && !accepted.is_empty() {
break;
@@ -6752,7 +6925,7 @@ where
if request.create_table {
if !target_table_preexisting {
let ddl = generate_create_table_ddl(
let ddl = generate_create_table_ddl_with_column_quoting(
&sql_target_columns,
&target_table,
&request.source_schema,
@@ -6761,18 +6934,20 @@ where
source_db_type,
None,
request.target_catalog.as_deref(),
request.quote_target_column_names,
);
let target_table_created = transfer_create_table_created(
execute_on_pool(state, target_pool_key, &ddl).await.map(|_| ()),
&format!("Failed to create table from MongoDB collection '{table}'"),
)?;
if target_table_created {
for stmt in generate_comment_ddl(
for stmt in generate_comment_ddl_with_column_quoting(
&sql_target_columns,
&target_table,
&request.target_schema,
target_db_type,
None,
request.quote_target_column_names,
) {
if let Err(e) = execute_on_pool(state, target_pool_key, &stmt).await {
log::warn!(
@@ -6817,7 +6992,7 @@ where
} else {
mongo_documents_to_rows(&documents, &sql_target_column_names)
};
let write_statements = generate_transfer_write_sql_batches(
let write_statements = generate_transfer_write_sql_batches_with_column_quoting(
&TransferMode::Append,
&sql_target_column_names,
&sql_target_column_types,
@@ -6829,6 +7004,7 @@ where
request.target_catalog.as_deref(),
false,
false,
request.quote_target_column_names,
)?;
for (statement_index, batch_sql) in write_statements.iter().enumerate() {
execute_on_pool(state, target_pool_key, batch_sql).await.map_err(|e| {
@@ -7156,7 +7332,8 @@ where
source_driver_profile.as_deref(),
target_driver_profile.as_deref(),
preserves_target_table_name,
);
) && (request.quote_target_column_names
|| !matches!(target_db_type, DatabaseType::Gaussdb | DatabaseType::OpenGauss));
let mut reused_source_ddl = false;
let ddl = if can_reuse_source_ddl {
let (source_ddl, source_ddl_was_read) = if let Some(catalog) =
@@ -7179,7 +7356,7 @@ where
Err(err) => {
log::warn!("[transfer] catalog DDL read failed for {table} in catalog '{catalog}': {err}; falling back to generated DDL");
(
generate_create_table_ddl(
generate_create_table_ddl_with_column_quoting(
&columns,
&target_table,
&request.source_schema,
@@ -7188,6 +7365,7 @@ where
source_db_type,
table_comment.as_deref(),
request.target_catalog.as_deref(),
request.quote_target_column_names,
),
false,
)
@@ -7206,7 +7384,7 @@ where
{
Ok(ddl) => (ddl, true),
Err(_) => (
generate_create_table_ddl(
generate_create_table_ddl_with_column_quoting(
&columns,
&target_table,
&request.source_schema,
@@ -7215,6 +7393,7 @@ where
source_db_type,
table_comment.as_deref(),
request.target_catalog.as_deref(),
request.quote_target_column_names,
),
false,
),
@@ -7223,7 +7402,7 @@ where
if contains_oceanbase_mysql_table_options(&source_ddl)
&& !db::oceanbase_mysql::is_profile(target_db_type, target_driver_profile.as_deref())
{
generate_create_table_ddl(
generate_create_table_ddl_with_column_quoting(
&columns,
&target_table,
&request.source_schema,
@@ -7232,6 +7411,7 @@ where
source_db_type,
table_comment.as_deref(),
request.target_catalog.as_deref(),
request.quote_target_column_names,
)
} else {
reused_source_ddl = source_ddl_was_read;
@@ -7244,7 +7424,7 @@ where
)
}
} else {
generate_create_table_ddl(
generate_create_table_ddl_with_column_quoting(
&columns,
&target_table,
&request.source_schema,
@@ -7253,6 +7433,7 @@ where
source_db_type,
table_comment.as_deref(),
request.target_catalog.as_deref(),
request.quote_target_column_names,
)
};
// MySQL-family targets: create the bare table first and add any foreign
@@ -7313,12 +7494,13 @@ where
if target_table_created {
pending_fk_alters
.extend(deferred_fk_alters.into_iter().map(|statement| (target_table.clone(), statement)));
let comment_stmts = generate_comment_ddl(
let comment_stmts = generate_comment_ddl_with_column_quoting(
&columns,
&target_table,
&request.target_schema,
target_db_type,
table_comment.as_deref(),
request.quote_target_column_names,
);
for stmt in &comment_stmts {
if let Err(e) = execute_on_pool(state, target_pool_key, stmt).await {
@@ -7391,7 +7573,12 @@ where
// can't accept the planned insert, fail fast here instead of truncating
// the target's existing data and then hitting an opaque driver error.
if request.create_table && target_table_preexisting {
let missing = missing_transfer_target_columns(&target_columns, &col_names, target_db_type);
let missing = missing_transfer_target_columns(
&target_columns,
&col_names,
target_db_type,
request.quote_target_column_names,
);
if !missing.is_empty() {
return Err(format!(
"Target table '{target_table}' already exists with a different structure and is missing column(s) \
@@ -7401,7 +7588,12 @@ where
));
}
let required = required_unmapped_transfer_target_columns(&target_columns, &col_names, target_db_type);
let required = required_unmapped_transfer_target_columns(
&target_columns,
&col_names,
target_db_type,
request.quote_target_column_names,
);
if !required.is_empty() {
return Err(format!(
"Target table '{target_table}' already exists with a different structure and has required column(s) \
@@ -7513,7 +7705,7 @@ where
break;
}
let write_statements = generate_transfer_write_sql_batches(
let write_statements = generate_transfer_write_sql_batches_with_column_quoting(
&effective_mode,
&col_names,
&col_types,
@@ -7525,6 +7717,7 @@ where
request.target_catalog.as_deref(),
overrides_postgres_system_values,
mysql_spatial_markers,
request.quote_target_column_names,
)?;
for (statement_index, batch_sql) in write_statements.iter().enumerate() {
execute_transfer_write_statement(
@@ -8323,6 +8516,7 @@ mod tests {
.unwrap();
assert_eq!(request.content, TransferContent::StructureAndData);
assert!(request.objects.is_empty());
assert!(request.quote_target_column_names);
}
#[test]
@@ -8346,6 +8540,7 @@ mod tests {
}],
mode: TransferMode::Append,
target_table_name_case: TransferTableNameCase::Preserve,
quote_target_column_names: true,
ownership_policy: TransferOwnershipPolicy::Preserve,
batch_size: 1000,
};
@@ -8353,6 +8548,7 @@ mod tests {
assert_eq!(json["content"], "structureOnly");
assert_eq!(json["objects"][0]["objectType"], "VIEW");
assert_eq!(json["objects"][0]["names"][0], "v1");
assert_eq!(json["quoteTargetColumnNames"], true);
}
mod transfer_family_tests {
@@ -8417,6 +8613,7 @@ mod tests {
create_table: true,
mode: TransferMode::Append,
target_table_name_case: TransferTableNameCase::Preserve,
quote_target_column_names: true,
ownership_policy: TransferOwnershipPolicy::Preserve,
batch_size: 1000,
content: TransferContent::DataOnly,
@@ -9415,6 +9612,7 @@ mod tests {
objects: Vec::new(),
mode: TransferMode::Append,
target_table_name_case: TransferTableNameCase::Preserve,
quote_target_column_names: true,
ownership_policy: TransferOwnershipPolicy::Preserve,
batch_size: 1000,
}
@@ -9637,7 +9835,7 @@ mod tests {
let col_names = vec!["id".to_string(), "name".to_string(), "extra_col".to_string()];
assert_eq!(
missing_transfer_target_columns(&target_columns, &col_names, &DatabaseType::Mysql),
missing_transfer_target_columns(&target_columns, &col_names, &DatabaseType::Mysql, true),
vec!["extra_col".to_string()]
);
}
@@ -9647,14 +9845,29 @@ mod tests {
let target_columns = vec![test_column("ID", "int"), test_column("Name", "varchar(32)")];
let col_names = vec!["id".to_string(), "name".to_string()];
assert!(missing_transfer_target_columns(&target_columns, &col_names, &DatabaseType::Mysql).is_empty());
assert!(missing_transfer_target_columns(&target_columns, &col_names, &DatabaseType::Kyuubi).is_empty());
assert!(missing_transfer_target_columns(&target_columns, &col_names, &DatabaseType::Mysql, true).is_empty());
assert!(missing_transfer_target_columns(&target_columns, &col_names, &DatabaseType::Kyuubi, true).is_empty());
assert_eq!(
missing_transfer_target_columns(&target_columns, &col_names, &DatabaseType::Postgres),
missing_transfer_target_columns(&target_columns, &col_names, &DatabaseType::Postgres, true),
vec!["id".to_string(), "name".to_string()]
);
}
#[test]
fn gauss_target_column_validation_ignores_case_when_quoting_disabled() {
let target_columns = vec![test_column("id", "int"), test_column("user_id", "bigint")];
let col_names = vec!["id".to_string(), "USER_ID".to_string()];
assert!(missing_transfer_target_columns(&target_columns, &col_names, &DatabaseType::Gaussdb, false).is_empty());
assert!(
missing_transfer_target_columns(&target_columns, &col_names, &DatabaseType::OpenGauss, false).is_empty()
);
assert_eq!(
missing_transfer_target_columns(&target_columns, &col_names, &DatabaseType::Gaussdb, true),
vec!["USER_ID".to_string()]
);
}
#[test]
fn transfer_target_column_validation_allows_omittable_target_columns() {
let target_columns = vec![
@@ -9684,11 +9897,15 @@ mod tests {
];
let col_names = vec!["id".to_string()];
assert!(required_unmapped_transfer_target_columns(&target_columns[..6], &col_names, &DatabaseType::Mysql)
assert!(required_unmapped_transfer_target_columns(
&target_columns[..6],
&col_names,
&DatabaseType::Mysql,
true
)
.is_empty());
assert!(required_unmapped_transfer_target_columns(&target_columns, &col_names, &DatabaseType::SqlServer, true)
.is_empty());
assert!(
required_unmapped_transfer_target_columns(&target_columns, &col_names, &DatabaseType::SqlServer).is_empty()
);
}
#[test]
@@ -9700,7 +9917,7 @@ mod tests {
let col_names = vec!["id".to_string()];
assert_eq!(
required_unmapped_transfer_target_columns(&target_columns, &col_names, &DatabaseType::Mysql),
required_unmapped_transfer_target_columns(&target_columns, &col_names, &DatabaseType::Mysql, true),
vec!["required_code".to_string()]
);
}
@@ -9753,6 +9970,78 @@ mod tests {
assert!(ddl.contains("PRIMARY KEY (`id`)"));
}
#[test]
fn gaussdb_transfer_can_leave_safe_target_columns_unquoted() {
let columns = vec![
db::ColumnInfo { is_primary_key: true, ..test_column("USER_ID", "int") },
test_column("CamelName", "varchar(32)"),
test_column("select", "varchar(32)"),
test_column("has space", "varchar(32)"),
];
let ddl = generate_create_table_ddl_with_column_quoting(
&columns,
"case_target",
"dbx_test",
"public",
&DatabaseType::Gaussdb,
&DatabaseType::Mysql,
None,
None,
false,
);
assert!(ddl.contains("USER_ID INTEGER"), "ddl: {ddl}");
assert!(ddl.contains("CamelName VARCHAR(32)"), "ddl: {ddl}");
assert!(ddl.contains("\"select\" VARCHAR(32)"), "ddl: {ddl}");
assert!(ddl.contains("\"has space\" VARCHAR(32)"), "ddl: {ddl}");
assert!(ddl.contains("PRIMARY KEY (USER_ID)"), "ddl: {ddl}");
let quoted = generate_create_table_ddl_with_column_quoting(
&columns,
"case_target",
"dbx_test",
"public",
&DatabaseType::Gaussdb,
&DatabaseType::Mysql,
None,
None,
true,
);
assert!(quoted.contains("\"USER_ID\" INTEGER"), "ddl: {quoted}");
assert!(quoted.contains("\"CamelName\" VARCHAR(32)"), "ddl: {quoted}");
}
#[test]
fn gaussdb_transfer_writes_use_the_same_target_column_quoting_policy() {
let columns = vec!["USER_ID".to_string(), "CamelName".to_string(), "select".to_string()];
let column_types = vec![Some("int".to_string()), Some("varchar".to_string()), Some("varchar".to_string())];
let rows = vec![vec![serde_json::json!(1), serde_json::json!("Alice"), serde_json::json!("ok")]];
let statements = generate_transfer_write_sql_batches_with_column_quoting(
&TransferMode::Append,
&columns,
&column_types,
&rows,
"case_target",
"public",
&DatabaseType::Gaussdb,
&[],
None,
false,
false,
false,
)
.unwrap();
assert_eq!(statements.len(), 1);
assert!(
statements[0].starts_with("INSERT INTO \"public\".\"case_target\" (USER_ID, CamelName, \"select\") VALUES"),
"sql: {}",
statements[0]
);
}
#[test]
fn oracle_to_mysql_create_table_ddl_strips_char_length_units() {
// Issue #6479 end-to-end: Oracle columns reported with CHAR length
crates/dbx-core/tests/live_mysql_transfer.rs
+5
View File
@@ -88,6 +88,7 @@ fn transfer_request(
objects: Vec::new(),
mode,
target_table_name_case: TransferTableNameCase::Preserve,
quote_target_column_names: true,
ownership_policy: TransferOwnershipPolicy::Preserve,
batch_size: 2,
}
@@ -167,6 +168,7 @@ async fn live_mysql_transfer_downgrades_unsupported_source_collations() {
objects: Vec::new(),
mode: TransferMode::Append,
target_table_name_case: TransferTableNameCase::Preserve,
quote_target_column_names: true,
ownership_policy: TransferOwnershipPolicy::Preserve,
batch_size: 10,
};
@@ -227,6 +229,7 @@ async fn live_mysql_transfer_downgrades_unsupported_source_collations() {
objects: Vec::new(),
mode: TransferMode::Append,
target_table_name_case: TransferTableNameCase::Preserve,
quote_target_column_names: true,
ownership_policy: TransferOwnershipPolicy::Preserve,
batch_size: 10,
};
@@ -527,6 +530,7 @@ async fn live_mysql_transfer_structure_overwrite_rejects_incompatible_target_col
objects: Vec::new(),
mode: TransferMode::Overwrite,
target_table_name_case: TransferTableNameCase::Preserve,
quote_target_column_names: true,
ownership_policy: TransferOwnershipPolicy::Preserve,
batch_size: 10,
};
@@ -579,6 +583,7 @@ async fn live_mysql_transfer_structure_overwrite_rejects_incompatible_target_col
objects: Vec::new(),
mode: TransferMode::Overwrite,
target_table_name_case: TransferTableNameCase::Preserve,
quote_target_column_names: true,
ownership_policy: TransferOwnershipPolicy::Preserve,
batch_size: 10,
};
crates/dbx-core/tests/live_postgres_transfer.rs
+5
View File
@@ -227,6 +227,7 @@ async fn live_postgres_transfer_upserts_generated_always_identity_values() {
objects: Vec::new(),
mode: TransferMode::Upsert,
target_table_name_case: TransferTableNameCase::Preserve,
quote_target_column_names: true,
ownership_policy: TransferOwnershipPolicy::Preserve,
batch_size: 100,
};
@@ -390,6 +391,7 @@ async fn live_postgres_structure_only_preserves_table_indexes() {
objects: Vec::new(),
mode: TransferMode::Append,
target_table_name_case: TransferTableNameCase::Preserve,
quote_target_column_names: true,
ownership_policy: TransferOwnershipPolicy::Preserve,
batch_size: 100,
};
@@ -634,6 +636,7 @@ async fn live_postgres_transfer_preserves_data_and_schema_objects() {
objects: Vec::new(),
mode: TransferMode::Append,
target_table_name_case: TransferTableNameCase::Preserve,
quote_target_column_names: true,
ownership_policy: TransferOwnershipPolicy::Preserve,
batch_size: 100,
};
@@ -921,6 +924,7 @@ async fn live_postgres_transfer_skips_create_ddl_for_existing_target_table() {
objects: Vec::new(),
mode: TransferMode::Append,
target_table_name_case: TransferTableNameCase::Preserve,
quote_target_column_names: true,
ownership_policy: TransferOwnershipPolicy::Preserve,
batch_size: 100,
};
@@ -1043,6 +1047,7 @@ async fn live_postgres_transfer_creates_selected_sequence_before_referencing_tab
],
mode: TransferMode::Append,
target_table_name_case: TransferTableNameCase::Preserve,
quote_target_column_names: true,
ownership_policy: TransferOwnershipPolicy::Preserve,
batch_size: 100,
};
crates/dbx-core/tests/live_sqlserver_completion.rs
+1
View File
@@ -1376,6 +1376,7 @@ async fn live_sqlserver_transfer_table_skips_rowversion_insert_column() {
objects: Vec::new(),
mode: dbx_core::transfer::TransferMode::Append,
target_table_name_case: dbx_core::transfer::TransferTableNameCase::Upper,
quote_target_column_names: true,
ownership_policy: dbx_core::transfer::TransferOwnershipPolicy::Preserve,
batch_size: 100,
};