From 42d9ad9fc20573ea6a19bf3dc87af1c9eebaef77 Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Thu, 13 Aug 2026 17:53:53 +0000 Subject: [PATCH] fix(gbase8s): use Informix SQL for table exports Closes #3584 --- crates/dbx-core/src/table_export.rs | 208 +++++++++++++++++++++------- crates/dbx-core/src/transfer.rs | 92 ++++++++++++ 2 files changed, 249 insertions(+), 51 deletions(-) diff --git a/crates/dbx-core/src/table_export.rs b/crates/dbx-core/src/table_export.rs index 0d38aa5c5d..b658b63261 100644 --- a/crates/dbx-core/src/table_export.rs +++ b/crates/dbx-core/src/table_export.rs @@ -232,10 +232,32 @@ fn format_markdown_rows(rows: &[Vec]) -> 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], 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], 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::>() - .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::>() + .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], primary_keys: &[String], @@ -445,7 +468,7 @@ async fn execute_external_driver_export_page( result_session_id: Option, cancel_token: CancellationToken, ) -> Result { - 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], 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], primary_keys: &[String], @@ -721,7 +745,7 @@ async fn fetch_paginated_table_export_batch( ) -> Result { 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], column_extras: &[Option], @@ -848,7 +873,7 @@ async fn try_export_native_table_stream( cancelled: Arc, cancel_token: CancellationToken, ) -> Result { - 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, 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, ®ular_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 { diff --git a/crates/dbx-core/src/transfer.rs b/crates/dbx-core/src/transfer.rs index 709987e8c9..d6c65d6e41 100644 --- a/crates/dbx-core/src/transfer.rs +++ b/crates/dbx-core/src/transfer.rs @@ -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(