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

feat(backup): support table filters and MySQL compatibility

Browse Source
This commit is contained in:
CSXFanMeng authored and GitHub committed 2026-07-19 10:32:33 +08:00
1 parent 09c3f7f44c
commit 7abac0959f
15 files changed
+386 -32
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/dbx-core/src/database_export.rs
+43 -15
View File
@@ -31,6 +31,8 @@ pub struct DatabaseExportRequest {
pub file_path: String,
#[serde(default)]
pub selected_tables: Vec<String>,
#[serde(default)]
pub excluded_tables: Vec<String>,
pub include_structure: bool,
pub include_data: bool,
pub include_objects: bool,
@@ -891,6 +893,7 @@ async fn list_postgres_export_sequences(
pool_key: &str,
schema: &str,
selected_tables: &[String],
excluded_tables: &[String],
include_objects: bool,
fail_on_error: bool,
) -> Result<Vec<PostgresExportSequence>, String> {
@@ -932,6 +935,7 @@ async fn list_postgres_export_sequences(
.map_err(|e| e.to_string())?;
let selected: HashSet<&str> = selected_tables.iter().map(String::as_str).collect();
let excluded: HashSet<&str> = excluded_tables.iter().map(String::as_str).collect();
let mut sequences = rows
.iter()
.map(|row| PostgresExportSequence {
@@ -948,8 +952,9 @@ async fn list_postgres_export_sequences(
owner_column: row.get::<_, Option<String>>(9),
})
.filter(|sequence| {
selected.is_empty()
|| sequence.owner_table.as_deref().map(|owner_table| selected.contains(owner_table)).unwrap_or(false)
sequence.owner_table.as_deref().is_none_or(|owner_table| !excluded.contains(owner_table))
&& (selected.is_empty()
|| sequence.owner_table.as_deref().is_some_and(|owner_table| selected.contains(owner_table)))
})
.filter(|sequence| sequence.owner_table.is_some() || (include_objects && selected.is_empty()))
.collect::<Vec<_>>();
@@ -1139,7 +1144,7 @@ pub async fn export_database_sql_core(
None,
)
.await?;
let all_tables = filter_selected_table_infos(all_tables, &request.selected_tables);
let all_tables = filter_export_table_infos(all_tables, &request.selected_tables, &request.excluded_tables);
// 4. Create file
let mut file = std::fs::File::create(&request.file_path).map_err(|e| format!("Failed to write file: {e}"))?;
@@ -1165,6 +1170,7 @@ pub async fn export_database_sql_core(
&pool_key,
&request.schema,
&request.selected_tables,
&request.excluded_tables,
request.include_objects,
request.fail_on_error,
)
@@ -1783,15 +1789,18 @@ pub async fn export_database_sql_core(
Ok(())
}
fn filter_selected_table_infos(
fn filter_export_table_infos(
tables: Vec<crate::types::TableInfo>,
selected_tables: &[String],
excluded_tables: &[String],
) -> Vec<crate::types::TableInfo> {
if selected_tables.is_empty() {
return tables;
}
let selected: HashSet<&str> = selected_tables.iter().map(String::as_str).collect();
tables.into_iter().filter(|table| selected.contains(table.name.as_str())).collect()
let excluded: HashSet<&str> = excluded_tables.iter().map(String::as_str).collect();
tables
.into_iter()
.filter(|table| selected.is_empty() || selected.contains(table.name.as_str()))
.filter(|table| !excluded.contains(table.name.as_str()))
.collect()
}
fn drop_table_if_exists_sql(table_name: &str, schema: &str, db_type: &DatabaseType) -> String {
@@ -1802,11 +1811,11 @@ fn drop_table_if_exists_sql(table_name: &str, schema: &str, db_type: &DatabaseTy
mod tests {
use super::concurrent_metadata_prefetch_allowed;
use super::{
build_database_sql_export, build_export_insert_statements, drop_table_if_exists_sql,
filter_selected_table_infos, format_export_sql_literal, generate_postgres_sequence_create_ddl,
generate_postgres_sequence_owner_ddl, generate_postgres_sequence_setval_sql, normalize_export_table_ddl,
record_export_error, BuildDatabaseSqlExportOptions, BuildExportInsertStatementsOptions, ExportedTableSql,
PostgresExportSequence, DATABASE_EXPORT_INSERT_BATCH_SIZE, DATABASE_EXPORT_ROW_LIMIT,
build_database_sql_export, build_export_insert_statements, drop_table_if_exists_sql, filter_export_table_infos,
format_export_sql_literal, generate_postgres_sequence_create_ddl, generate_postgres_sequence_owner_ddl,
generate_postgres_sequence_setval_sql, normalize_export_table_ddl, record_export_error,
BuildDatabaseSqlExportOptions, BuildExportInsertStatementsOptions, ExportedTableSql, PostgresExportSequence,
DATABASE_EXPORT_INSERT_BATCH_SIZE, DATABASE_EXPORT_ROW_LIMIT,
};
use crate::models::connection::DatabaseType;
use crate::types::TableInfo;
@@ -1848,7 +1857,7 @@ mod tests {
fn filters_export_tables_by_selected_names() {
let tables = vec![table("users", "TABLE"), table("orders", "TABLE"), table("active_users", "VIEW")];
let filtered = filter_selected_table_infos(tables, &["active_users".to_string(), "users".to_string()]);
let filtered = filter_export_table_infos(tables, &["active_users".to_string(), "users".to_string()], &[]);
assert_eq!(filtered.iter().map(|table| table.name.as_str()).collect::<Vec<_>>(), vec!["users", "active_users"]);
}
@@ -1857,11 +1866,30 @@ mod tests {
fn keeps_all_export_tables_when_selection_is_empty() {
let tables = vec![table("users", "TABLE"), table("orders", "TABLE")];
let filtered = filter_selected_table_infos(tables.clone(), &[]);
let filtered = filter_export_table_infos(tables.clone(), &[], &[]);
assert_eq!(filtered.iter().map(|table| table.name.as_str()).collect::<Vec<_>>(), vec!["users", "orders"]);
}
#[test]
fn excludes_export_tables_by_name() {
let tables = vec![table("users", "TABLE"), table("audit_log", "TABLE"), table("active_users", "VIEW")];
let filtered = filter_export_table_infos(tables, &[], &["audit_log".to_string(), "active_users".to_string()]);
assert_eq!(filtered.iter().map(|table| table.name.as_str()).collect::<Vec<_>>(), vec!["users"]);
}
#[test]
fn exclusions_take_precedence_over_selected_tables() {
let tables = vec![table("users", "TABLE"), table("orders", "TABLE")];
let filtered =
filter_export_table_infos(tables, &["users".to_string(), "orders".to_string()], &["orders".to_string()]);
assert_eq!(filtered.iter().map(|table| table.name.as_str()).collect::<Vec<_>>(), vec!["users"]);
}
#[test]
fn builds_drop_table_if_exists_with_qualified_mysql_name() {
let sql = drop_table_if_exists_sql("users", "app", &DatabaseType::Mysql);
crates/dbx-core/src/query.rs
+63 -7
View File
@@ -2918,14 +2918,26 @@ fn postgres_transaction_begin_sql(consistent_snapshot: bool) -> &'static str {
}
}
fn mysql_transaction_begin_sql(consistent_snapshot: bool) -> &'static str {
fn mysql_transaction_begin_sql_candidates(consistent_snapshot: bool) -> &'static [&'static str] {
if consistent_snapshot {
"START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY"
&[
"START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY",
"START TRANSACTION WITH CONSISTENT SNAPSHOT",
"START TRANSACTION",
]
} else {
"START TRANSACTION"
&["START TRANSACTION"]
}
}
fn mysql_transaction_isolation_sql(consistent_snapshot: bool) -> Option<&'static str> {
consistent_snapshot.then_some("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
}
fn mysql_error_is_syntax_error(error: &mysql_async::Error) -> bool {
matches!(error, mysql_async::Error::Server(server_error) if server_error.code == 1064)
}
async fn begin_transaction_session(
state: &AppState,
connection_id: &str,
@@ -2968,8 +2980,25 @@ async fn begin_transaction_session(
}
TxnPoolHandle::Mysql(mysql_pool) => {
let mut conn = mysql_pool.get_conn().await.map_err(|e| format!("Failed to get MySQL connection: {e}"))?;
let begin_sql = mysql_transaction_begin_sql(consistent_snapshot);
conn.query_drop(begin_sql).await.map_err(|e| format!("START TRANSACTION failed: {e}"))?;
if let Some(isolation_sql) = mysql_transaction_isolation_sql(consistent_snapshot) {
conn.query_drop(isolation_sql).await.map_err(|e| format!("SET TRANSACTION failed: {e}"))?;
}
let mut syntax_errors = Vec::new();
for begin_sql in mysql_transaction_begin_sql_candidates(consistent_snapshot) {
match conn.query_drop(*begin_sql).await {
Ok(()) => {
syntax_errors.clear();
break;
}
Err(error) if mysql_error_is_syntax_error(&error) => {
syntax_errors.push(format!("{begin_sql}: {error}"));
}
Err(error) => return Err(format!("START TRANSACTION failed: {error}")),
}
}
if !syntax_errors.is_empty() {
return Err(format!("START TRANSACTION failed for all compatible forms: {}", syntax_errors.join("; ")));
}
TxnConnection::Mysql(conn)
}
};
@@ -4956,8 +4985,35 @@ mod tests {
}
#[test]
fn database_backup_transactions_request_consistent_read_only_snapshots() {
fn database_backup_transactions_request_consistent_snapshots() {
assert_eq!(postgres_transaction_begin_sql(true), "BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY");
assert_eq!(mysql_transaction_begin_sql(true), "START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY");
assert_eq!(mysql_transaction_isolation_sql(true), Some("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ"));
assert_eq!(
mysql_transaction_begin_sql_candidates(true),
[
"START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY",
"START TRANSACTION WITH CONSISTENT SNAPSHOT",
"START TRANSACTION",
]
);
assert_eq!(mysql_transaction_isolation_sql(false), None);
assert_eq!(mysql_transaction_begin_sql_candidates(false), ["START TRANSACTION"]);
}
#[test]
fn mysql_backup_transaction_only_falls_back_for_syntax_errors() {
let syntax_error = mysql_async::Error::Server(mysql_async::ServerError {
code: 1064,
message: "unsupported transaction characteristic".to_string(),
state: "42000".to_string(),
});
let permission_error = mysql_async::Error::Server(mysql_async::ServerError {
code: 1044,
message: "access denied".to_string(),
state: "42000".to_string(),
});
assert!(mysql_error_is_syntax_error(&syntax_error));
assert!(!mysql_error_is_syntax_error(&permission_error));
}
}
crates/dbx-core/tests/database_export_prefetch.rs
+1
View File
@@ -172,6 +172,7 @@ async fn database_export_writes_structure_and_data_for_all_tables() {
schema: "public".to_string(),
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: false,