github/dbx
4
0
Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-10-20 00:17:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(export): include Oracle sequences in database export

Browse Source
This commit is contained in:
Jeff
2026-10-12 15:22:12 +08:00
committed by GitHub
parent 524a93203b
commit 860d901115
2 changed files with 288 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/dbx-core/src/data/database_export.rs
+107 -2
View File
@@ -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$$_<object id>` 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<crate::types::ObjectInfo> = Vec::new();
let mut triggers: Vec<crate::types::ObjectInfo> = Vec::new();
let mut events: Vec<crate::types::ObjectInfo> = Vec::new();
let mut oracle_sequences: Vec<crate::types::ObjectInfo> = 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')";
crates/dbx-core/tests/live_oracle_database_export.rs
+181
View File
@@ -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::<u16>().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();
}