From 860d901115cd7dc11fc2f09e4573392b0246dcad Mon Sep 17 00:00:00 2001 From: Jeff Date: Mon, 12 Oct 2026 15:22:12 +0800 Subject: [PATCH] fix(export): include Oracle sequences in database export --- crates/dbx-core/src/data/database_export.rs | 109 ++++++++++- .../tests/live_oracle_database_export.rs | 181 ++++++++++++++++++ 2 files changed, 288 insertions(+), 2 deletions(-) create mode 100644 crates/dbx-core/tests/live_oracle_database_export.rs diff --git a/crates/dbx-core/src/data/database_export.rs b/crates/dbx-core/src/data/database_export.rs index 5f919d00b0..074256743b 100644 --- a/crates/dbx-core/src/data/database_export.rs +++ b/crates/dbx-core/src/data/database_export.rs @@ -264,6 +264,21 @@ fn exports_mysql_trigger_objects(db_type: DatabaseType) -> bool { matches!(db_type, DatabaseType::Mysql) } +/// Oracle writes each table's triggers into its table DDL (see `loadTableTriggerDDLs` in +/// `agents/drivers/oracle-go`), so a trigger that reads `seq.NEXTVAL` reaches the script +/// whenever structure is exported. The standalone sequences it depends on are written +/// before the tables of a whole-schema structure export so the script replays (#11673). +fn exports_oracle_sequence_objects(request: &DatabaseExportRequest, db_type: DatabaseType) -> bool { + matches!(db_type, DatabaseType::Oracle) && request.include_structure && request.selected_tables.is_empty() +} + +/// Oracle backs every identity column with a system-named `ISEQ$$_` sequence. +/// The identity clause in the table DDL recreates it, so exporting it would only add an +/// orphan sequence to the target schema. +fn is_oracle_identity_sequence(name: &str) -> bool { + name.starts_with("ISEQ$$_") +} + fn database_export_total_objects(request: &DatabaseExportRequest, counts: &DatabaseExportObjectCounts) -> usize { let mut total = 0; if exports_database_tables(request) { @@ -4008,8 +4023,11 @@ async fn export_database_sql_core_inner( let mut functions: Vec = Vec::new(); let mut triggers: Vec = Vec::new(); let mut events: Vec = Vec::new(); + let mut oracle_sequences: Vec = Vec::new(); + let collects_schema_wide_objects = exports_schema_wide_objects(request); + let collects_oracle_sequences = exports_oracle_sequence_objects(request, db_type); - if exports_schema_wide_objects(request) { + if collects_schema_wide_objects || collects_oracle_sequences { match crate::schema::list_objects_core( state, &request.connection_id, @@ -4029,6 +4047,15 @@ async fn export_database_sql_core_inner( if is_postgres_extension_member_routine(obj, &postgres_extension_members) { continue; } + if ot == "SEQUENCE" { + if collects_oracle_sequences && !is_oracle_identity_sequence(&obj.name) { + oracle_sequences.push(obj.clone()); + } + continue; + } + if !collects_schema_wide_objects { + continue; + } if ot.contains("PROCEDURE") { procedures.push(obj.clone()); } else if ot.contains("FUNCTION") { @@ -4052,7 +4079,8 @@ async fn export_database_sql_core_inner( &DatabaseExportObjectCounts { tables: tables.len(), views: views.len(), - sequences: postgres_sequences.iter().filter(|sequence| !sequence.created_by_table).count(), + sequences: postgres_sequences.iter().filter(|sequence| !sequence.created_by_table).count() + + oracle_sequences.len(), extensions: postgres_extensions.len(), enums: postgres_enums.len(), procedures: procedures.len(), @@ -4257,6 +4285,57 @@ async fn export_database_sql_core_inner( object_index += 1; } + for sequence in &oracle_sequences { + if is_export_cancelled(&request.export_id).await { + return Err("Export cancelled".to_string()); + } + + emit_database_export_running( + &on_progress, + &request.export_id, + sequence.name.clone(), + object_index, + total_objects, + total_rows_exported, + false, + ); + + match crate::schema::get_object_source_core( + state, + &request.connection_id, + &request.database, + &request.schema, + &sequence.name, + crate::db::ObjectSourceKind::Sequence, + None, + None, + ) + .await + { + Ok(obj_source) => { + let source = build_database_export_object_source_sql( + db_type, + &ObjectSourceKind::Sequence, + &sequence.name, + &obj_source.source, + request.drop_table_if_exists, + ); + if !source.is_empty() { + writeln!(file, "{source}\n").map_err(|e| format!("Failed to write file: {e}"))?; + } + } + Err(e) => { + record_export_error( + &mut file, + request.fail_on_error, + format!("exporting sequence {}: {e}", sequence.name), + &mut lenient_errors, + )?; + } + } + object_index += 1; + } + for (table_index, table_info) in tables.iter().enumerate().filter(|_| exports_database_tables(request)) { // Check cancellation if is_export_cancelled(&request.export_id).await { @@ -5327,6 +5406,32 @@ mod tests { assert_eq!(database_export_total_objects(&request, &counts), 5); } + #[test] + fn oracle_sequences_are_exported_with_whole_schema_structure() { + use super::{exports_oracle_sequence_objects, is_oracle_identity_sequence}; + + let cases = [ + ("structure", DatabaseType::Oracle, export_request(true, false, false, Vec::new()), true), + ("all", DatabaseType::Oracle, export_request(true, true, true, Vec::new()), true), + ("data only", DatabaseType::Oracle, export_request(false, true, true, Vec::new()), false), + ( + "selected tables", + DatabaseType::Oracle, + export_request(true, true, true, vec!["ORDERS".to_string()]), + false, + ), + ("postgres", DatabaseType::Postgres, export_request(true, true, true, Vec::new()), false), + ("oceanbase oracle", DatabaseType::OceanbaseOracle, export_request(true, true, true, Vec::new()), false), + ]; + for (name, db_type, request, expected) in cases { + assert_eq!(exports_oracle_sequence_objects(&request, db_type), expected, "{name}"); + } + + assert!(is_oracle_identity_sequence("ISEQ$$_115126")); + assert!(!is_oracle_identity_sequence("ORDER_SEQ")); + assert!(!is_oracle_identity_sequence("iseq$$_custom")); + } + #[test] fn mysql_database_export_drops_triggers_and_events_before_create() { let trigger = "CREATE DEFINER=`root`@`%` TRIGGER `trg_orders_ai` AFTER INSERT ON `orders` FOR EACH ROW INSERT INTO audit_log(msg) VALUES ('x')"; diff --git a/crates/dbx-core/tests/live_oracle_database_export.rs b/crates/dbx-core/tests/live_oracle_database_export.rs new file mode 100644 index 0000000000..eac7f53bac --- /dev/null +++ b/crates/dbx-core/tests/live_oracle_database_export.rs @@ -0,0 +1,181 @@ +use dbx_core::connection::AppState; +use dbx_core::database_export::{export_database_sql_core, DatabaseExportRequest}; +use dbx_core::models::connection::{ConnectionConfig, DatabaseType}; +use dbx_core::query::execute_sql_statement; +use dbx_core::sql::SqlFileRequest; +use dbx_core::sql_file_import::execute_sql_file_path; +use std::sync::Arc; +use tokio_util::sync::CancellationToken; + +fn live_oracle_config(id: &str) -> ConnectionConfig { + let host = std::env::var("DBX_LIVE_ORACLE_HOST").expect("DBX_LIVE_ORACLE_HOST"); + let port = std::env::var("DBX_LIVE_ORACLE_PORT").ok().and_then(|value| value.parse::().ok()).unwrap_or(1521); + let service = std::env::var("DBX_LIVE_ORACLE_SERVICE").expect("DBX_LIVE_ORACLE_SERVICE"); + let username = std::env::var("DBX_LIVE_ORACLE_USER").expect("DBX_LIVE_ORACLE_USER"); + let password = std::env::var("DBX_LIVE_ORACLE_PASSWORD").expect("DBX_LIVE_ORACLE_PASSWORD"); + + serde_json::from_value(serde_json::json!({ + "id": id, + "name": id, + "db_type": DatabaseType::Oracle, + "host": host, + "port": port, + "username": username, + "password": password, + "database": service, + "connect_timeout_secs": 10, + "query_timeout_secs": 60, + "idle_timeout_secs": 60, + "keepalive_interval_secs": 0 + })) + .expect("live Oracle config should deserialize") +} + +fn first_cell(result: &dbx_core::types::QueryResult) -> String { + match result.rows.first().and_then(|row| row.first()) { + Some(serde_json::Value::String(value)) => value.clone(), + Some(value) => value.to_string(), + None => String::new(), + } +} + +/// The export covers the whole schema and the replay recreates every object in it, +/// so `DBX_LIVE_ORACLE_USER` must own an otherwise empty, disposable schema. +#[tokio::test] +#[ignore = "requires a disposable Oracle schema"] +async fn live_oracle_database_export_replays_trigger_sequences() { + let suffix = uuid::Uuid::new_v4().simple().to_string()[..8].to_uppercase(); + let connection_id = format!("live-oracle-export-{suffix}"); + let dir = std::env::temp_dir().join(format!("dbx-live-oracle-export-{suffix}")); + std::fs::create_dir_all(&dir).unwrap(); + let storage = dbx_core::persistence::test_storage::open(&dir.join("storage.db")).await.unwrap(); + let state = Arc::new(AppState::new(storage)); + state.configs.write().await.insert(connection_id.clone(), live_oracle_config(&connection_id)); + let service = std::env::var("DBX_LIVE_ORACLE_SERVICE").unwrap(); + let schema = std::env::var("DBX_LIVE_ORACLE_USER").unwrap().to_uppercase(); + + let orders = format!("DBX_EXPORT_ORDERS_{suffix}"); + let sequence = format!("DBX_EXPORT_SEQ_{suffix}"); + let trigger = format!("DBX_EXPORT_TRG_{suffix}"); + let identity = format!("DBX_EXPORT_IDENT_{suffix}"); + let drop_all = [ + format!("DROP TABLE {orders} PURGE"), + format!("DROP TABLE {identity} PURGE"), + format!("DROP SEQUENCE {sequence}"), + ]; + let run = |sql: String| { + let state = state.clone(); + let connection_id = connection_id.clone(); + let service = service.clone(); + let schema = schema.clone(); + async move { execute_sql_statement(&state, &connection_id, &service, &sql, Some(&schema), None).await } + }; + + let file_path = dir.join("export.sql"); + let outcome = async { + for sql in [ + format!("CREATE TABLE {orders} (ID NUMBER PRIMARY KEY, NOTE VARCHAR2(50))"), + format!("CREATE SEQUENCE {sequence} START WITH 100 INCREMENT BY 5 NOCACHE"), + format!( + "CREATE OR REPLACE TRIGGER {trigger} BEFORE INSERT ON {orders} FOR EACH ROW\nBEGIN\n IF :NEW.ID IS NULL THEN SELECT {sequence}.NEXTVAL INTO :NEW.ID FROM DUAL; END IF;\nEND;" + ), + format!("CREATE TABLE {identity} (ID NUMBER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY)"), + format!("INSERT INTO {orders} (NOTE) VALUES ('first')"), + format!("INSERT INTO {orders} (NOTE) VALUES ('second')"), + ] { + run(sql).await?; + } + + let request = DatabaseExportRequest { + export_id: format!("live-oracle-export-{suffix}"), + connection_id: connection_id.clone(), + database: service.clone(), + schema: schema.clone(), + file_path: file_path.to_string_lossy().to_string(), + selected_tables: Vec::new(), + excluded_tables: Vec::new(), + include_structure: true, + include_data: true, + include_objects: true, + include_create_database: false, + drop_table_if_exists: false, + omit_auto_increment: false, + preserve_original_language: false, + fail_on_error: true, + prevent_overwrite: false, + output_compression: Default::default(), + insert_dialect: Default::default(), + insert_mode: Default::default(), + snapshot_session_id: None, + batch_size: 1000, + split_max_mb: None, + }; + export_database_sql_core(&state, &request, |_| {}).await?; + let exported = std::fs::read_to_string(&file_path).map_err(|error| error.to_string())?; + let sequence_position = exported + .find(&format!("CREATE SEQUENCE \"{schema}\".\"{sequence}\"")) + .ok_or_else(|| format!("CREATE SEQUENCE {sequence} missing from export:\n{exported}"))?; + let trigger_position = exported + .find(&format!("TRIGGER \"{schema}\".\"{trigger}\"")) + .ok_or_else(|| format!("trigger {trigger} missing from export:\n{exported}"))?; + if sequence_position > trigger_position { + return Err(format!("sequence {sequence} is exported after the trigger that uses it")); + } + if exported.contains("ISEQ$$_") { + return Err("identity column sequence was exported as a standalone sequence".to_string()); + } + + for sql in drop_all.clone() { + run(sql).await?; + } + let import_request = SqlFileRequest { + txn_session_id: None, + execution_id: format!("live-oracle-import-{suffix}"), + connection_id: connection_id.clone(), + database: service.clone(), + schema: Some(schema.clone()), + file_path: file_path.to_string_lossy().to_string(), + continue_on_error: false, + selected_tables: None, + part_cooldown_ms: 0, + skip_relational_constraints: false, + failure_log_enabled: true, + }; + execute_sql_file_path( + &state, + &import_request, + &file_path, + CancellationToken::new(), + std::time::Instant::now(), + |_| {}, + ) + .await?; + + let sequences = run(format!("SELECT COUNT(*) FROM USER_SEQUENCES WHERE SEQUENCE_NAME = '{sequence}'")).await?; + if first_cell(&sequences) != "1" { + return Err(format!("{sequence} was not recreated")); + } + let trigger_status = run(format!( + "SELECT o.STATUS || '/' || t.STATUS FROM USER_OBJECTS o JOIN USER_TRIGGERS t ON t.TRIGGER_NAME = o.OBJECT_NAME \ + WHERE o.OBJECT_TYPE = 'TRIGGER' AND o.OBJECT_NAME = '{trigger}'" + )) + .await?; + if first_cell(&trigger_status) != "VALID/ENABLED" { + return Err(format!("{trigger} replayed as {}", first_cell(&trigger_status))); + } + run(format!("INSERT INTO {orders} (NOTE) VALUES ('after replay')")).await?; + let next_id = run(format!("SELECT ID FROM {orders} WHERE NOTE = 'after replay'")).await?; + if first_cell(&next_id) != "110" { + return Err(format!("{sequence}.NEXTVAL resumed at {} instead of 110", first_cell(&next_id))); + } + Ok::<_, String>(()) + } + .await; + + for sql in drop_all { + let _ = run(sql).await; + } + drop(state); + std::fs::remove_dir_all(&dir).unwrap(); + outcome.unwrap(); +}