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

fix(gbase8s): use Informix SQL for table exports

Browse Source 
Closes #3584
This commit is contained in:
t8y2 committed 2026-08-13 17:53:53 +00:00
1 parent b119ca36ae
commit 42d9ad9fc2
2 files changed
+249 -51
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/dbx-core/src/table_export.rs
+157 -51
View File
@@ -232,10 +232,32 @@ fn format_markdown_rows(rows: &[Vec<Value>]) -> String {
.join("\n")
}
#[derive(Clone, Copy, Debug, PartialEq)]
struct TableExportSqlContext<'a> {
database_type: DatabaseType,
schema: Option<&'a str>,
}
fn table_export_sql_context<'a>(
database_type: DatabaseType,
driver_profile: Option<&str>,
schema: Option<&'a str>,
) -> TableExportSqlContext<'a> {
// GBase8s selects from the active database and rejects the owner-qualified
// names produced for MySQL-family GBase connections.
if database_type == DatabaseType::Gbase
&& driver_profile.is_some_and(|profile| profile.eq_ignore_ascii_case("gbase8s"))
{
TableExportSqlContext { database_type: DatabaseType::Informix, schema: None }
} else {
TableExportSqlContext { database_type, schema }
}
}
#[allow(clippy::too_many_arguments)]
fn table_page_sql(
request: &TableExportRequest,
db_type: &DatabaseType,
sql_context: &TableExportSqlContext<'_>,
col_names: &[String],
column_types: &[Option<String>],
primary_keys: &[String],
@@ -248,8 +270,8 @@ fn table_page_sql(
keyset_pagination_sql_with_identifier_quote(
col_names,
&request.table_name,
request.schema.as_deref().unwrap_or(""),
db_type,
sql_context.schema.unwrap_or(""),
&sql_context.database_type,
primary_keys,
last_pk_values,
batch_size,
@@ -259,8 +281,8 @@ fn table_page_sql(
pagination_sql_with_filter_order_and_identifier_quote(
col_names,
&request.table_name,
request.schema.as_deref().unwrap_or(""),
db_type,
sql_context.schema.unwrap_or(""),
&sql_context.database_type,
offset,
batch_size,
request.where_input.as_deref(),
@@ -269,7 +291,7 @@ fn table_page_sql(
request.identifier_quote.as_deref(),
)
};
replace_mysql_spatial_export_select_list(sql, request, db_type, col_names, column_types)
replace_mysql_spatial_export_select_list(sql, request, &sql_context.database_type, col_names, column_types)
}
fn mysql_spatial_export_column_expression(column: &str, identifier_quote: Option<&str>) -> String {
@@ -342,30 +364,31 @@ fn replace_mysql_spatial_export_select_list(
fn table_cursor_sql(
request: &TableExportRequest,
db_type: &DatabaseType,
sql_context: &TableExportSqlContext<'_>,
col_names: &[String],
column_types: &[Option<String>],
primary_keys: &[String],
) -> String {
let full_table = crate::sql_dialect::table_data_qualified_table_name(
Some(*db_type),
request.schema.as_deref(),
Some(sql_context.database_type),
sql_context.schema,
&request.table_name,
request.identifier_quote.as_deref(),
);
let col_list = mysql_spatial_export_select_list(request, db_type, col_names, column_types).unwrap_or_else(|| {
col_names
.iter()
.map(|column| {
crate::sql_dialect::quote_table_data_identifier(
Some(*db_type),
column,
request.identifier_quote.as_deref(),
)
})
.collect::<Vec<_>>()
.join(", ")
});
let col_list = mysql_spatial_export_select_list(request, &sql_context.database_type, col_names, column_types)
.unwrap_or_else(|| {
col_names
.iter()
.map(|column| {
crate::sql_dialect::quote_table_data_identifier(
Some(sql_context.database_type),
column,
request.identifier_quote.as_deref(),
)
})
.collect::<Vec<_>>()
.join(", ")
});
let predicate = crate::sql_dialect::normalize_where_input(request.where_input.as_deref());
let where_clause = if predicate.is_empty() { String::new() } else { format!(" WHERE ({predicate})") };
let order_by = request
@@ -385,7 +408,7 @@ fn table_cursor_sql(
format!(
"{} ASC",
crate::sql_dialect::quote_table_data_identifier(
Some(*db_type),
Some(sql_context.database_type),
column,
request.identifier_quote.as_deref(),
)
@@ -437,7 +460,7 @@ async fn execute_external_driver_export_page(
state: &AppState,
pool_key: &str,
request: &TableExportRequest,
db_type: &DatabaseType,
sql_context: &TableExportSqlContext<'_>,
col_names: &[String],
column_types: &[Option<String>],
primary_keys: &[String],
@@ -445,7 +468,7 @@ async fn execute_external_driver_export_page(
result_session_id: Option<String>,
cancel_token: CancellationToken,
) -> Result<QueryResult, String> {
let sql = table_cursor_sql(request, db_type, col_names, column_types, primary_keys);
let sql = table_cursor_sql(request, sql_context, col_names, column_types, primary_keys);
let max_rows = request.row_limit.unwrap_or(i32::MAX as usize).min(i32::MAX as usize).max(1);
let timeout_secs = table_export_query_timeout_secs(state, pool_key).await;
execute_sql_statement_with_options(
@@ -504,6 +527,7 @@ async fn fetch_table_export_batch(
pool_key: &str,
request: &TableExportRequest,
db_type: &DatabaseType,
sql_context: &TableExportSqlContext<'_>,
col_names: &[String],
column_types: &[Option<String>],
primary_keys: &[String],
@@ -561,7 +585,7 @@ async fn fetch_table_export_batch(
match table_export_cursor_kind(state, pool_key).await {
Some(TableExportCursorKind::Agent) => {
*table_read_attempted = true;
let sql = table_cursor_sql(request, db_type, col_names, column_types, primary_keys);
let sql = table_cursor_sql(request, sql_context, col_names, column_types, primary_keys);
let max_rows = request.row_limit.unwrap_or(i32::MAX as usize);
let query_timeout = table_export_query_timeout_secs(state, pool_key).await;
let params = AgentTableReadStartParams {
@@ -600,7 +624,7 @@ async fn fetch_table_export_batch(
state,
pool_key,
request,
db_type,
sql_context,
col_names,
column_types,
primary_keys,
@@ -656,7 +680,7 @@ async fn fetch_table_export_batch(
state,
pool_key,
request,
db_type,
sql_context,
col_names,
column_types,
primary_keys,
@@ -693,7 +717,7 @@ async fn fetch_table_export_batch(
state,
pool_key,
request,
db_type,
sql_context,
col_names,
column_types,
primary_keys,
@@ -710,7 +734,7 @@ async fn fetch_paginated_table_export_batch(
state: &AppState,
pool_key: &str,
request: &TableExportRequest,
db_type: &DatabaseType,
sql_context: &TableExportSqlContext<'_>,
col_names: &[String],
column_types: &[Option<String>],
primary_keys: &[String],
@@ -721,7 +745,7 @@ async fn fetch_paginated_table_export_batch(
) -> Result<QueryResult, String> {
let sql = table_page_sql(
request,
db_type,
sql_context,
col_names,
column_types,
primary_keys,
@@ -837,6 +861,7 @@ async fn try_export_native_table_stream(
pool_key: &str,
request: &TableExportRequest,
db_type: &DatabaseType,
sql_context: &TableExportSqlContext<'_>,
col_names: &[String],
column_types: &[Option<String>],
column_extras: &[Option<String>],
@@ -848,7 +873,7 @@ async fn try_export_native_table_stream(
cancelled: Arc<AtomicBool>,
cancel_token: CancellationToken,
) -> Result<bool, String> {
let sql = table_cursor_sql(request, db_type, col_names, column_types, primary_keys);
let sql = table_cursor_sql(request, sql_context, col_names, column_types, primary_keys);
let mut rows_exported = 0_u64;
let progress_interval = batch_size.max(1) as u64;
@@ -1259,14 +1284,15 @@ async fn export_table_data_core_inner(
cancelled: Arc<AtomicBool>,
cancel_token: CancellationToken,
) -> Result<(), String> {
// 1. Get database type
let db_type = state
.configs
.read()
.await
.get(&request.connection_id)
.map(|c| c.db_type)
.ok_or_else(|| format!("Connection config not found: {}", request.connection_id))?;
// 1. Get database type and the profile that can refine its SQL dialect.
let (db_type, driver_profile) = {
let configs = state.configs.read().await;
let config = configs
.get(&request.connection_id)
.ok_or_else(|| format!("Connection config not found: {}", request.connection_id))?;
(config.db_type, config.driver_profile.clone())
};
let sql_context = table_export_sql_context(db_type, driver_profile.as_deref(), request.schema.as_deref());
// 2. Get pool
let client_session_id = table_export_client_session_id(&request.export_id);
@@ -1346,8 +1372,8 @@ async fn export_table_data_core_inner(
} else {
let count_query = count_sql_with_where_and_identifier_quote(
&request.table_name,
request.schema.as_deref().unwrap_or(""),
&db_type,
sql_context.schema.unwrap_or(""),
&sql_context.database_type,
request.where_input.as_deref(),
None,
request.identifier_quote.as_deref(),
@@ -1382,6 +1408,7 @@ async fn export_table_data_core_inner(
&pool_key,
request,
&db_type,
&sql_context,
&col_names,
&column_types,
&column_extras,
@@ -1442,6 +1469,7 @@ async fn export_table_data_core_inner(
&pool_key,
request,
&db_type,
&sql_context,
&col_names,
&column_types,
&primary_keys,
@@ -1531,6 +1559,7 @@ async fn export_table_data_core_inner(
&pool_key,
request,
&db_type,
&sql_context,
&col_names,
&column_types,
&primary_keys,
@@ -1631,6 +1660,7 @@ async fn export_table_data_core_inner(
&pool_key,
request,
&db_type,
&sql_context,
&col_names,
&column_types,
&primary_keys,
@@ -1725,6 +1755,7 @@ async fn export_table_data_core_inner(
&pool_key,
request,
&db_type,
&sql_context,
&col_names,
&column_types,
&primary_keys,
@@ -1808,6 +1839,7 @@ async fn export_table_data_core_inner(
&pool_key,
request,
&db_type,
&sql_context,
&col_names,
&column_types,
&primary_keys,
@@ -1890,6 +1922,7 @@ async fn export_table_data_core_inner(
&pool_key,
request,
&db_type,
&sql_context,
&col_names,
&column_types,
&primary_keys,
@@ -2254,10 +2287,11 @@ mod tests {
numeric_column_right_align: false,
column_comments: None,
};
let context = table_export_sql_context(DatabaseType::Oracle, None, request.schema.as_deref());
let sql = table_cursor_sql(
&request,
&DatabaseType::Oracle,
&context,
&[String::from("id"), String::from("status")],
&[],
&[String::from("id")],
@@ -2272,6 +2306,76 @@ mod tests {
assert!(!sql.contains("ROWNUM"));
}
#[test]
fn gbase8s_table_export_uses_owner_free_informix_queries() {
let request = TableExportRequest {
export_id: "export-gbase8s".to_string(),
connection_id: "conn-1".to_string(),
database: "appdb".to_string(),
schema: Some("gbasedbt".to_string()),
identifier_quote: Some(String::new()),
table_name: "orders".to_string(),
file_path: "orders.txt".to_string(),
format: "txt".to_string(),
columns: None,
column_types: None,
primary_keys: None,
where_input: None,
order_by: None,
skip_count: false,
batch_size: Some(50),
row_limit: None,
date_time_format: None,
numeric_column_right_align: false,
column_comments: None,
};
let columns = vec!["id".to_string(), "payload".to_string()];
let primary_keys = vec!["id".to_string()];
let context = table_export_sql_context(DatabaseType::Gbase, Some("gbase8s"), request.schema.as_deref());
assert_eq!(context.database_type, DatabaseType::Informix);
assert_eq!(context.schema, None);
assert_eq!(
table_cursor_sql(&request, &context, &columns, &[], &primary_keys),
"SELECT id, payload FROM orders ORDER BY id ASC"
);
assert_eq!(
table_page_sql(&request, &context, &columns, &[], &primary_keys, false, &[], 100, 50),
"SELECT SKIP 100 FIRST 50 id, payload FROM orders ORDER BY id"
);
assert_eq!(
table_page_sql(&request, &context, &columns, &[], &primary_keys, true, &[json!(10)], 0, 50),
"SELECT FIRST 50 id, payload FROM orders WHERE id > 10 ORDER BY id ASC"
);
assert_eq!(
count_sql_with_where_and_identifier_quote(
&request.table_name,
context.schema.unwrap_or(""),
&context.database_type,
None,
None,
request.identifier_quote.as_deref(),
),
"SELECT COUNT(*) FROM orders"
);
let regular_gbase = table_export_sql_context(DatabaseType::Gbase, Some("gbase8a"), request.schema.as_deref());
assert_eq!(regular_gbase.database_type, DatabaseType::Gbase);
assert_eq!(regular_gbase.schema, Some("gbasedbt"));
assert_eq!(
table_page_sql(&request, &regular_gbase, &columns, &[], &primary_keys, false, &[], 100, 50),
"SELECT \"id\", \"payload\" FROM \"gbasedbt\".\"orders\" ORDER BY \"id\" LIMIT 50 OFFSET 100"
);
let informix = table_export_sql_context(DatabaseType::Informix, None, request.schema.as_deref());
assert_eq!(informix.database_type, DatabaseType::Informix);
assert_eq!(informix.schema, Some("gbasedbt"));
assert_eq!(
table_page_sql(&request, &informix, &columns, &[], &primary_keys, false, &[], 0, 50),
"SELECT FIRST 50 id, payload FROM gbasedbt.orders ORDER BY id"
);
}
#[test]
fn gaussdb_m_table_export_uses_backticks_across_all_query_paths() {
let request = TableExportRequest {
@@ -2297,17 +2401,18 @@ mod tests {
};
let columns = vec!["id".to_string(), "DisplayName".to_string()];
let primary_keys = vec!["id".to_string()];
let context = table_export_sql_context(DatabaseType::Gaussdb, None, request.schema.as_deref());
assert_eq!(
table_cursor_sql(&request, &DatabaseType::Gaussdb, &columns, &[], &primary_keys),
table_cursor_sql(&request, &context, &columns, &[], &primary_keys),
"SELECT id, `DisplayName` FROM app_schema.`order` ORDER BY id ASC"
);
assert_eq!(
table_page_sql(&request, &DatabaseType::Gaussdb, &columns, &[], &primary_keys, false, &[], 100, 100),
table_page_sql(&request, &context, &columns, &[], &primary_keys, false, &[], 100, 100),
"SELECT id, `DisplayName` FROM app_schema.`order` ORDER BY id LIMIT 100 OFFSET 100"
);
assert_eq!(
table_page_sql(&request, &DatabaseType::Gaussdb, &columns, &[], &primary_keys, true, &[json!(10)], 0, 100,),
table_page_sql(&request, &context, &columns, &[], &primary_keys, true, &[json!(10)], 0, 100,),
"SELECT id, `DisplayName` FROM app_schema.`order` WHERE id > 10 ORDER BY id ASC LIMIT 100"
);
assert_eq!(
@@ -2349,20 +2454,20 @@ mod tests {
let columns = vec!["id".to_string(), "geom".to_string(), "name".to_string()];
let column_types = vec![Some("int".to_string()), Some("geometry".to_string()), Some("varchar".to_string())];
let primary_keys = vec!["id".to_string()];
let context = table_export_sql_context(DatabaseType::Mysql, None, None);
let cursor_sql = table_cursor_sql(&request, &DatabaseType::Mysql, &columns, &column_types, &primary_keys);
let cursor_sql = table_cursor_sql(&request, &context, &columns, &column_types, &primary_keys);
assert!(cursor_sql.contains("ST_SRID(`geom`), ':', HEX(ST_AsWKB(`geom`))"));
assert!(cursor_sql.contains("AS `geom`"));
assert!(!cursor_sql.contains("SELECT `id`, `geom`, `name`"));
let page_sql =
table_page_sql(&request, &DatabaseType::Mysql, &columns, &column_types, &primary_keys, false, &[], 0, 100);
let page_sql = table_page_sql(&request, &context, &columns, &column_types, &primary_keys, false, &[], 0, 100);
assert!(page_sql.contains("ST_AsWKB(`geom`)"));
assert!(page_sql.contains("LIMIT 100 OFFSET 0"));
let mut csv_request = request;
csv_request.format = "csv".to_string();
let csv_sql = table_cursor_sql(&csv_request, &DatabaseType::Mysql, &columns, &column_types, &primary_keys);
let csv_sql = table_cursor_sql(&csv_request, &context, &columns, &column_types, &primary_keys);
assert_eq!(csv_sql, "SELECT `id`, `geom`, `name` FROM `spatial_data` ORDER BY `id` ASC");
}
@@ -2400,7 +2505,8 @@ mod tests {
numeric_column_right_align: false,
column_comments: None,
};
let sql = table_cursor_sql(&request, &DatabaseType::Oracle, &columns, &[], &primary_keys);
let context = table_export_sql_context(DatabaseType::Oracle, None, request.schema.as_deref());
let sql = table_cursor_sql(&request, &context, &columns, &[], &primary_keys);
assert_eq!(sql, "SELECT \"ID\", \"NAME\" FROM \"APP\".\"USERS\"");
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
crates/dbx-core/src/transfer.rs
+92
View File
@@ -3657,6 +3657,13 @@ pub fn pagination_sql(
let base_sql = format!("SELECT {col_list} FROM {full_table}");
oracle_rownum_page_sql(&col_list, base_sql, offset, limit)
}
DatabaseType::Informix => {
if offset == 0 {
format!("SELECT FIRST {limit} {col_list} FROM {full_table}")
} else {
format!("SELECT SKIP {offset} FIRST {limit} {col_list} FROM {full_table}")
}
}
DatabaseType::SqlServer | DatabaseType::Dameng => {
format!(
"SELECT {col_list} FROM {full_table} ORDER BY (SELECT NULL) OFFSET {offset} ROWS FETCH NEXT {limit} ROWS ONLY"
@@ -3692,6 +3699,14 @@ pub fn pagination_sql_with_order(
let base_sql = format!("SELECT {col_list} FROM {full_table}{order_by}");
oracle_rownum_page_sql(&col_list, base_sql, offset, limit)
}
DatabaseType::Informix => {
let order_by = order_expression.map(|value| format!(" ORDER BY {value}")).unwrap_or_default();
if offset == 0 {
format!("SELECT FIRST {limit} {col_list} FROM {full_table}{order_by}")
} else {
format!("SELECT SKIP {offset} FIRST {limit} {col_list} FROM {full_table}{order_by}")
}
}
DatabaseType::SqlServer | DatabaseType::Dameng => {
let order_by = order_expression.unwrap_or_else(|| "(SELECT NULL)".to_string());
format!(
@@ -3768,6 +3783,14 @@ pub fn pagination_sql_with_filter_order_and_identifier_quote(
let base_sql = format!("SELECT {col_list} FROM {full_table}{where_clause}{order_by}");
oracle_rownum_page_sql(&col_list, base_sql, offset, limit)
}
DatabaseType::Informix => {
let order_by = order_expression.map(|value| format!(" ORDER BY {value}")).unwrap_or_default();
if offset == 0 {
format!("SELECT FIRST {limit} {col_list} FROM {full_table}{where_clause}{order_by}")
} else {
format!("SELECT SKIP {offset} FIRST {limit} {col_list} FROM {full_table}{where_clause}{order_by}")
}
}
DatabaseType::SqlServer | DatabaseType::Dameng => {
let order_by = order_expression.unwrap_or_else(|| "(SELECT NULL)".to_string());
format!(
@@ -3865,6 +3888,9 @@ pub fn keyset_pagination_sql_with_identifier_quote(
let base_sql = format!("SELECT {col_list} FROM {full_table}{where_clause} ORDER BY {order}");
oracle_rownum_page_sql(&col_list, base_sql, 0, limit)
}
DatabaseType::Informix => {
format!("SELECT FIRST {limit} {col_list} FROM {full_table}{where_clause} ORDER BY {order}")
}
DatabaseType::SqlServer | DatabaseType::Dameng => {
format!(
"SELECT {col_list} FROM {full_table}{where_clause} ORDER BY {order} OFFSET 0 ROWS FETCH NEXT {limit} ROWS ONLY"
@@ -9284,6 +9310,72 @@ mod tests {
assert_eq!(sql, "SELECT `id`, `name` FROM `users` ORDER BY `id` LIMIT 200, 300");
}
#[test]
fn informix_pagination_uses_first_and_optional_skip() {
let columns = [String::from("id"), String::from("name")];
assert_eq!(
pagination_sql(&columns, "users", "app", &DatabaseType::Informix, 0, 100),
"SELECT FIRST 100 \"id\", \"name\" FROM \"app\".\"users\""
);
assert_eq!(
pagination_sql(&columns, "users", "app", &DatabaseType::Informix, 200, 100),
"SELECT SKIP 200 FIRST 100 \"id\", \"name\" FROM \"app\".\"users\""
);
}
#[test]
fn informix_ordered_pagination_uses_first_and_optional_skip() {
let columns = [String::from("id"), String::from("name")];
let order = [String::from("id")];
assert_eq!(
pagination_sql_with_order(&columns, "users", "app", &DatabaseType::Informix, 0, 100, &order, None),
"SELECT FIRST 100 \"id\", \"name\" FROM \"app\".\"users\" ORDER BY id"
);
assert_eq!(
pagination_sql_with_order(&columns, "users", "app", &DatabaseType::Informix, 200, 100, &order, None),
"SELECT SKIP 200 FIRST 100 \"id\", \"name\" FROM \"app\".\"users\" ORDER BY id"
);
}
#[test]
fn informix_filtered_pagination_preserves_filter_and_order() {
let columns = [String::from("id"), String::from("status")];
let default_order = [String::from("id")];
assert_eq!(
pagination_sql_with_filter_order(
&columns,
"users",
"app",
&DatabaseType::Informix,
200,
100,
Some("WHERE status = 'active'"),
Some("id DESC"),
&default_order,
),
"SELECT SKIP 200 FIRST 100 id, status FROM \"app\".\"users\" WHERE (status = 'active') ORDER BY id DESC"
);
}
#[test]
fn informix_keyset_pagination_uses_first() {
assert_eq!(
keyset_pagination_sql(
&[String::from("id"), String::from("name")],
"users",
"app",
&DatabaseType::Informix,
&[String::from("id")],
&[json!(25)],
100,
),
"SELECT FIRST 100 id, name FROM \"app\".\"users\" WHERE id > 25 ORDER BY id ASC"
);
}
#[test]
fn dameng_export_pagination_uses_offset_fetch() {
let sql = pagination_sql(