Files
dbx/crates/dbx-core/src/database_export.rs
T

4464 lines
177 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::{HashMap, HashSet};
use std::io::{BufWriter, Write};
use std::sync::RwLock;
use crate::connection::task_client_session_id;
use crate::models::connection::DatabaseType;
use crate::mysql_ddl_normalize::DdlNormalizeOptions;
use crate::object_source_sql::build_export_object_source_sql;
use crate::sql_dialect::{qualified_table_name, uses_single_row_insert_statements};
use crate::transfer::{
format_ch_array_sql_literal, format_pg_array_sql_literal, is_identity_column_extra,
is_mysql_generated_column_extra, keyset_pagination_sql_with_identifier_quote, quote_identifier,
quote_postgres_string_literal, wrap_dameng_identity_insert_sql_for_table,
};
use crate::types::ObjectSourceKind;
static EXPORT_CANCELLED: std::sync::LazyLock<RwLock<HashSet<String>>> =
std::sync::LazyLock::new(|| RwLock::new(HashSet::new()));
pub fn database_export_client_session_id(export_id: &str) -> String {
task_client_session_id("database-export", export_id)
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DatabaseExportRequest {
pub export_id: String,
pub connection_id: String,
pub database: String,
pub schema: String,
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,
/// Include a MySQL `CREATE DATABASE IF NOT EXISTS` and `USE` preamble.
/// The option is intentionally opt-in because it needs CREATE privileges
/// when importing into a new server.
#[serde(default)]
pub include_create_database: bool,
#[serde(default)]
pub drop_table_if_exists: bool,
/// Drop the table-level `AUTO_INCREMENT=N` clause from exported MySQL DDL,
/// so the script can initialize a fresh database without pinning a sequence
/// position. No-op for non-MySQL databases. Defaults to `false` (preserve).
#[serde(default)]
pub omit_auto_increment: bool,
#[serde(default)]
pub fail_on_error: bool,
#[serde(default)]
pub snapshot_session_id: Option<String>,
pub batch_size: usize,
}
#[derive(Debug, Default, PartialEq, Eq)]
struct DatabaseExportObjectCounts {
tables: usize,
views: usize,
sequences: usize,
extensions: usize,
procedures: usize,
functions: usize,
}
fn exports_database_tables(request: &DatabaseExportRequest) -> bool {
request.include_structure || request.include_data
}
fn exports_database_routines(request: &DatabaseExportRequest) -> bool {
// Routine export is schema-wide, so an explicit table selection must not
// add unrelated procedures or functions to either execution or progress.
request.include_objects && request.selected_tables.is_empty()
}
fn database_export_total_objects(request: &DatabaseExportRequest, counts: &DatabaseExportObjectCounts) -> usize {
let mut total = 0;
if exports_database_tables(request) {
total += counts.tables;
}
if request.include_structure {
total += counts.sequences + counts.extensions;
}
if request.include_objects {
total += counts.views;
}
if exports_database_routines(request) {
total += counts.procedures + counts.functions;
}
total
}
fn mysql_sql_string_literal(value: &str) -> String {
format!("'{}'", value.replace('\\', "\\\\").replace('\'', "''"))
}
fn mysql_view_dependencies_sql(database: &str) -> String {
let database = mysql_sql_string_literal(database);
format!(
"SELECT VIEW_NAME, TABLE_NAME FROM information_schema.VIEW_TABLE_USAGE \
WHERE VIEW_SCHEMA = {database} AND TABLE_SCHEMA = {database} \
ORDER BY VIEW_NAME, TABLE_NAME"
)
}
fn mysql_view_dependencies_from_rows(rows: &[Vec<Value>]) -> Vec<(String, String)> {
rows.iter()
.filter_map(|row| {
let view_name = row.first()?.as_str()?.trim();
let referenced_name = row.get(1)?.as_str()?.trim();
(!view_name.is_empty() && !referenced_name.is_empty())
.then(|| (view_name.to_string(), referenced_name.to_string()))
})
.collect()
}
async fn list_mysql_export_view_dependencies(
state: &crate::connection::AppState,
connection_id: &str,
database: &str,
) -> Result<Vec<(String, String)>, String> {
let result = crate::query::execute_sql_statement_with_options(
state,
connection_id,
database,
&mysql_view_dependencies_sql(database),
None,
None,
crate::query::QueryExecutionOptions { max_rows: Some(usize::MAX), ..Default::default() },
)
.await?;
Ok(mysql_view_dependencies_from_rows(&result.rows))
}
fn sort_export_views_by_dependencies<'a>(
views: &[&'a crate::types::TableInfo],
dependencies: &[(String, String)],
) -> Vec<&'a crate::types::TableInfo> {
let names = views.iter().map(|view| view.name.clone()).collect::<Vec<_>>();
let sorted_names = crate::transfer::sort_table_names_by_dependencies(&names, dependencies, true);
let views_by_name = views.iter().map(|view| (view.name.as_str(), *view)).collect::<HashMap<_, _>>();
sorted_names.iter().filter_map(|name| views_by_name.get(name.as_str()).copied()).collect()
}
fn mysql_database_export_preamble(database: &str, charset: Option<&str>, collation: Option<&str>) -> String {
let database = quote_identifier(database, &DatabaseType::Mysql);
let charset = charset.map(str::trim).filter(|value| !value.is_empty());
let collation = collation.map(str::trim).filter(|value| !value.is_empty());
let options = match charset {
Some(charset) => match collation {
Some(collation) => format!(" CHARACTER SET {charset} COLLATE {collation}"),
None => format!(" CHARACTER SET {charset}"),
},
None => String::new(),
};
format!("CREATE DATABASE IF NOT EXISTS {database}{options};\nUSE {database};\n")
}
async fn mysql_database_export_preamble_for_request(
state: &crate::connection::AppState,
request: &DatabaseExportRequest,
) -> String {
let metadata_sql = format!(
"SELECT DEFAULT_CHARACTER_SET_NAME, DEFAULT_COLLATION_NAME FROM information_schema.SCHEMATA WHERE SCHEMA_NAME = {}",
mysql_sql_string_literal(&request.database)
);
let metadata = crate::query::execute_sql_statement(
state,
&request.connection_id,
&request.database,
&metadata_sql,
None,
None,
)
.await
.ok()
.and_then(|result| {
result.rows.first().map(|row| {
(
row.first().and_then(Value::as_str).map(str::to_string),
row.get(1).and_then(Value::as_str).map(str::to_string),
)
})
});
match metadata {
Some((charset, collation)) => {
mysql_database_export_preamble(&request.database, charset.as_deref(), collation.as_deref())
}
None => mysql_database_export_preamble(&request.database, None, None),
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DatabaseBackupSnapshot {
pub session_id: String,
pub schemas: Vec<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ExportProgress {
pub export_id: String,
pub current_object: String,
pub object_index: usize,
pub total_objects: usize,
pub rows_exported: u64,
pub total_rows: Option<u64>,
pub status: ExportStatus,
pub error: Option<String>,
/// True while listing schema / prefetching table metadata — before objects are written.
#[serde(default)]
pub preparing: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ExportStatus {
Running,
Writing,
Done,
Error,
Cancelled,
}
pub const DATABASE_EXPORT_ROW_LIMIT: usize = 10_000;
pub const DATABASE_EXPORT_PAGE_SIZE: usize = 500;
pub const DATABASE_EXPORT_INSERT_BATCH_SIZE: usize = 100;
pub const DATABASE_EXPORT_TARGET_STATEMENT_BYTES: usize = 512 * 1024;
#[derive(Debug, Clone, PartialEq, Eq)]
struct PostgresExportSequence {
name: String,
data_type: String,
start_value: String,
min_value: String,
max_value: String,
increment: String,
cycle: bool,
cache_value: String,
last_value: Option<String>,
owner_table: Option<String>,
owner_column: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct PostgresExportExtension {
name: String,
schema: String,
}
#[derive(Debug, Default)]
struct PostgresExtensionMembers {
relation_names: HashSet<String>,
function_keys: HashSet<(String, String)>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ExportedTableSql {
pub display_name: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub database_type: Option<DatabaseType>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub identifier_quote: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub schema: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub table_name: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub qualified_table_name: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub ddl: Option<String>,
#[serde(default)]
pub columns: Vec<String>,
#[serde(default)]
pub column_types: Vec<Option<String>>,
#[serde(default)]
pub column_extras: Vec<Option<String>>,
#[serde(default)]
pub rows: Vec<Vec<Value>>,
#[serde(default)]
pub truncated: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BuildExportInsertStatementsOptions {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub database_type: Option<DatabaseType>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub identifier_quote: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub schema: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub table_name: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub qualified_table_name: Option<String>,
#[serde(default)]
pub columns: Vec<String>,
#[serde(default)]
pub column_types: Vec<Option<String>>,
#[serde(default)]
pub column_extras: Vec<Option<String>>,
#[serde(default)]
pub rows: Vec<Vec<Value>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub batch_size: Option<usize>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BuildExportSqlInsertOptions {
#[serde(flatten)]
pub insert: BuildExportInsertStatementsOptions,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BuildDatabaseSqlExportOptions {
pub database_name: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub exported_at: Option<String>,
#[serde(default)]
pub tables: Vec<ExportedTableSql>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub row_limit_per_table: Option<usize>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub insert_batch_size: Option<usize>,
/// Optional connection info for FK-aware table ordering.
/// When set, the caller should sort tables by dependency before passing them
/// to `build_database_sql_export`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub connection_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub database: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub schema: Option<String>,
/// Drop the table-level `AUTO_INCREMENT=N` clause from exported MySQL DDL.
/// Defaults to `false` (preserve). See `DatabaseExportRequest::omit_auto_increment`.
#[serde(default)]
pub omit_auto_increment: bool,
}
pub fn format_export_sql_literal(value: &Value) -> String {
format_export_sql_literal_for_database(value, None)
}
fn format_export_sql_literal_for_database(value: &Value, database_type: Option<DatabaseType>) -> String {
if value.is_null() {
return "NULL".to_string();
}
if let Some(number) = value.as_number() {
return number.to_string();
}
if let Some(value) = value.as_bool() {
if matches!(database_type, Some(DatabaseType::Dameng) | Some(DatabaseType::SqlServer)) {
return if value { "1" } else { "0" }.to_string();
}
return if value { "TRUE" } else { "FALSE" }.to_string();
}
if let Some(arr) = value.as_array() {
return format_pg_array_sql_literal(arr);
}
let text = value.as_str().map_or_else(|| value.to_string(), ToString::to_string);
quote_export_sql_string_for_database(&text, database_type)
}
fn format_export_sql_literal_typed(
value: &Value,
database_type: Option<DatabaseType>,
column_type: Option<&str>,
sqlserver_unicode_string: bool,
) -> String {
if is_postgres_json_export_column(database_type, column_type) {
return format_postgres_json_export_literal(value);
}
if is_postgres_vector_export_column(database_type, column_type) {
return format_postgres_vector_export_literal(value);
}
if matches!(database_type, Some(DatabaseType::Mysql)) && column_type.is_some_and(is_mysql_bit_type) {
return format_mysql_bit_literal(value);
}
if let Some(literal) = format_mysql_spatial_export_literal(value, database_type, column_type) {
return literal;
}
if is_mysql_compatible_export_literal_target(database_type) {
if column_type.is_some_and(is_mysql_binary_export_type) {
if let Some(literal) = format_mysql_binary_export_literal(value) {
return literal;
}
}
if column_type.is_some_and(is_export_numeric_type) {
if let Some(literal) = format_export_numeric_literal(value) {
return literal;
}
}
}
if let Some(arr) = value.as_array() {
if matches!(database_type, Some(DatabaseType::ClickHouse) | Some(DatabaseType::Databend)) {
return format_ch_array_sql_literal(arr);
}
}
if let Some(literal) = format_oracle_export_temporal_literal(value, database_type, column_type) {
return literal;
}
if let Some(literal) = format_export_temporal_literal(value, database_type, column_type) {
return literal;
}
if sqlserver_unicode_string {
if let Some(text) = value.as_str() {
return format!("N{}", quote_export_sql_string(text));
}
}
format_export_sql_literal_for_database(value, database_type)
}
fn format_postgres_json_export_literal(value: &Value) -> String {
if value.is_null() {
return "NULL".to_string();
}
let text = value.as_str().map_or_else(|| value.to_string(), ToString::to_string);
// PostgreSQL standard strings keep backslashes literal; JSON text needs its
// own escape sequences, so only SQL-escape the surrounding string delimiter.
quote_postgres_string_literal(&text)
}
fn format_postgres_vector_export_literal(value: &Value) -> String {
if value.is_null() {
return "NULL".to_string();
}
let text = match value {
// pgvector vector/halfvec are scalar extension types whose importable
// literal grammar uses square brackets, unlike PostgreSQL arrays.
Value::Array(arr) => format_postgres_vector_export_text(arr),
Value::String(text) => text.to_string(),
_ => value.to_string(),
};
quote_postgres_string_literal(&text)
}
fn format_postgres_vector_export_text(arr: &[Value]) -> String {
let elements = arr.iter().map(format_postgres_vector_export_element).collect::<Vec<_>>();
format!("[{}]", elements.join(","))
}
fn format_postgres_vector_export_element(value: &Value) -> String {
match value {
Value::String(text) => text.trim().to_string(),
Value::Number(number) => number.to_string(),
Value::Bool(value) => value.to_string(),
Value::Null => "NULL".to_string(),
_ => value.to_string(),
}
}
fn quote_export_sql_string(text: &str) -> String {
format!("'{}'", text.replace('\\', "\\\\").replace('\'', "''"))
}
fn is_sqlserver_unicode_export_type(column_type: &str) -> bool {
let base = column_type.trim().split(|ch: char| ch == '(' || ch.is_whitespace()).next().unwrap_or("");
["nchar", "nvarchar", "ntext", "sysname"].iter().any(|candidate| base.eq_ignore_ascii_case(candidate))
}
fn quote_export_sql_string_for_database(text: &str, database_type: Option<DatabaseType>) -> String {
match database_type {
Some(DatabaseType::Dameng) => quote_dameng_export_sql_string(text),
Some(DatabaseType::Postgres) => quote_postgres_string_literal(text),
database_type if is_mysql_compatible_export_literal_target(database_type) => {
quote_mysql_compatible_export_sql_string(text)
}
_ => quote_export_sql_string(text),
}
}
fn quote_dameng_export_sql_string(text: &str) -> String {
if !text.contains('\0') {
return quote_export_sql_string(text);
}
let mut parts = Vec::new();
for (index, segment) in text.split('\0').enumerate() {
if index > 0 {
parts.push("CHR(0)".to_string());
}
if !segment.is_empty() {
parts.push(quote_export_sql_string(segment));
}
}
parts.join(" || ")
}
fn quote_mysql_compatible_export_sql_string(text: &str) -> String {
format!("'{}'", escape_mysql_compatible_export_sql_string(text))
}
fn escape_mysql_compatible_export_sql_string(text: &str) -> String {
let mut escaped = String::with_capacity(text.len());
for ch in text.chars() {
match ch {
// MySQL-family dumps should keep control characters out of the
// physical script layout while relying on the dialect's escapes.
'\0' => escaped.push_str("\\0"),
'\x08' => escaped.push_str("\\b"),
'\n' => escaped.push_str("\\n"),
'\r' => escaped.push_str("\\r"),
'\t' => escaped.push_str("\\t"),
'\x0c' => escaped.push_str("\\f"),
'\x1a' => escaped.push_str("\\Z"),
'\\' => escaped.push_str("\\\\"),
'\'' => escaped.push_str("''"),
_ => escaped.push(ch),
}
}
escaped
}
fn is_mysql_compatible_export_literal_target(database_type: Option<DatabaseType>) -> bool {
matches!(
database_type,
Some(DatabaseType::Mysql | DatabaseType::Doris | DatabaseType::StarRocks | DatabaseType::Goldendb)
)
}
fn format_oracle_export_temporal_literal(
value: &Value,
database_type: Option<DatabaseType>,
column_type: Option<&str>,
) -> Option<String> {
if !matches!(database_type, Some(DatabaseType::Oracle | DatabaseType::OceanbaseOracle)) {
return None;
}
let column_type = column_type?;
let lower = column_type.trim().trim_matches('"').to_ascii_lowercase();
let base = lower.split(['(', ' ', '\t', '\n']).next().unwrap_or("");
let kind = match base {
"timestampdty" => ExportTemporalKind::DateTime,
"timestamptz_dty" => ExportTemporalKind::DateTimeWithTimeZone,
_ => export_temporal_column_kind(database_type, column_type)?,
};
let parts = parse_export_date_parts(value.as_str()?)?;
if base == "date" {
return Some(format_oracle_export_date_parts_literal(&parts));
}
let fraction = parts.fraction.as_deref().unwrap_or_default();
let datetime = format!("{} {}{fraction}", parts.date, parts.time);
let mask = if fraction.is_empty() { "YYYY-MM-DD HH24:MI:SS" } else { "YYYY-MM-DD HH24:MI:SS.FF" };
if kind == ExportTemporalKind::DateTimeWithTimeZone && !parts.zone.is_empty() {
let zone = normalize_export_timezone(&parts.zone);
return Some(format!("TO_TIMESTAMP_TZ('{datetime} {zone}', '{mask} TZH:TZM')"));
}
Some(format!("TO_TIMESTAMP('{datetime}', '{mask}')"))
}
fn format_oracle_export_date_parts_literal(parts: &ExportRfc3339Parts) -> String {
if export_temporal_parts_are_midnight(parts) {
format!("DATE '{}'", parts.date)
} else {
format!("TO_DATE('{} {}', 'YYYY-MM-DD HH24:MI:SS')", parts.date, parts.time)
}
}
fn export_temporal_parts_are_midnight(parts: &ExportRfc3339Parts) -> bool {
parts.time == "00:00:00"
&& parts
.fraction
.as_deref()
.map(|fraction| fraction.trim_start_matches('.').chars().all(|ch| ch == '0'))
.unwrap_or(true)
}
fn format_export_temporal_literal(
value: &Value,
database_type: Option<DatabaseType>,
column_type: Option<&str>,
) -> Option<String> {
let text = value.as_str()?;
let column_type = column_type?;
if database_type == Some(DatabaseType::SqlServer) {
return crate::sqlserver_temporal::normalize_sqlserver_temporal_literal(text, Some(column_type))
.map(|text| quote_export_sql_string(&text));
}
let kind = export_temporal_column_kind(database_type, column_type)?;
format_rfc3339_export_temporal_text(text, kind, database_type).map(|text| quote_export_sql_string(&text))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ExportTemporalKind {
Date,
Time,
DateTime,
DateTimeWithTimeZone,
}
fn export_temporal_column_kind(database_type: Option<DatabaseType>, column_type: &str) -> Option<ExportTemporalKind> {
let lower = column_type.trim().trim_matches('"').to_ascii_lowercase();
let base = lower.split(['(', ' ', '\t', '\n']).next().unwrap_or("");
match base {
"date" if matches!(database_type, Some(DatabaseType::Oracle | DatabaseType::OceanbaseOracle)) => {
Some(ExportTemporalKind::DateTime)
}
"date" => Some(ExportTemporalKind::Date),
"time" => Some(ExportTemporalKind::Time),
"datetime" | "datetime2" | "smalldatetime" | "datetime64" => Some(ExportTemporalKind::DateTime),
"datetimeoffset" | "timestamptz" => Some(ExportTemporalKind::DateTimeWithTimeZone),
_ if lower.starts_with("timestamp")
&& (lower.contains("with time zone") || lower.contains("with local time zone")) =>
{
Some(ExportTemporalKind::DateTimeWithTimeZone)
}
_ if lower.starts_with("timestamp") => Some(ExportTemporalKind::DateTime),
_ => None,
}
}
fn format_rfc3339_export_temporal_text(
text: &str,
kind: ExportTemporalKind,
database_type: Option<DatabaseType>,
) -> Option<String> {
let parts = parse_export_rfc3339_parts(text)?;
let fraction = normalize_export_fraction(parts.fraction.as_deref(), database_type);
match kind {
ExportTemporalKind::Date => Some(parts.date),
ExportTemporalKind::Time => Some(format!("{}{fraction}", parts.time)),
ExportTemporalKind::DateTime => Some(format!("{} {}{fraction}", parts.date, parts.time)),
ExportTemporalKind::DateTimeWithTimeZone => {
Some(format!("{} {}{fraction}{}", parts.date, parts.time, normalize_export_timezone(&parts.zone)))
}
}
}
struct ExportRfc3339Parts {
date: String,
time: String,
fraction: Option<String>,
zone: String,
}
fn parse_export_date_parts(text: &str) -> Option<ExportRfc3339Parts> {
parse_export_rfc3339_parts(text).or_else(|| parse_export_local_temporal_parts(text))
}
fn parse_export_local_temporal_parts(text: &str) -> Option<ExportRfc3339Parts> {
let bytes = text.as_bytes();
if bytes.len() < 10 || bytes.get(4) != Some(&b'-') || bytes.get(7) != Some(&b'-') {
return None;
}
let date = &text[0..10];
if !date.as_bytes().iter().enumerate().all(|(index, byte)| matches!(index, 4 | 7) || byte.is_ascii_digit()) {
return None;
}
if bytes.len() == 10 {
return Some(ExportRfc3339Parts {
date: date.to_string(),
time: "00:00:00".to_string(),
fraction: None,
zone: String::new(),
});
}
let separator = *bytes.get(10)?;
if separator != b'T' && separator != b' ' {
return None;
}
if bytes.len() < 19 || bytes.get(13) != Some(&b':') || bytes.get(16) != Some(&b':') {
return None;
}
let time = &text[11..19];
if !time.as_bytes().iter().enumerate().all(|(index, byte)| matches!(index, 2 | 5) || byte.is_ascii_digit()) {
return None;
}
let rest = &text[19..];
let fraction = if let Some(rest) = rest.strip_prefix('.') {
let digit_count = rest.chars().take_while(|ch| ch.is_ascii_digit()).count();
if digit_count == 0 || digit_count > 9 || digit_count != rest.len() {
return None;
}
Some(format!(".{}", &rest[..digit_count]))
} else if rest.is_empty() {
None
} else {
return None;
};
Some(ExportRfc3339Parts { date: date.to_string(), time: time.to_string(), fraction, zone: String::new() })
}
fn parse_export_rfc3339_parts(text: &str) -> Option<ExportRfc3339Parts> {
let bytes = text.as_bytes();
if bytes.len() < 20 || bytes.get(4) != Some(&b'-') || bytes.get(7) != Some(&b'-') {
return None;
}
let separator = *bytes.get(10)?;
if separator != b'T' && separator != b' ' {
return None;
}
if bytes.get(13) != Some(&b':') || bytes.get(16) != Some(&b':') {
return None;
}
let date = &text[0..10];
let time = &text[11..19];
if !date.as_bytes().iter().enumerate().all(|(index, byte)| matches!(index, 4 | 7) || byte.is_ascii_digit())
|| !time.as_bytes().iter().enumerate().all(|(index, byte)| matches!(index, 2 | 5) || byte.is_ascii_digit())
{
return None;
}
let rest = &text[19..];
let (fraction, zone) = if let Some(rest) = rest.strip_prefix('.') {
let digit_count = rest.chars().take_while(|ch| ch.is_ascii_digit()).count();
if digit_count == 0 || digit_count > 9 {
return None;
}
(Some(format!(".{}", &rest[..digit_count])), &rest[digit_count..])
} else {
(None, rest)
};
if zone.eq_ignore_ascii_case("z") || is_export_timezone_offset(zone) {
Some(ExportRfc3339Parts { date: date.to_string(), time: time.to_string(), fraction, zone: zone.to_string() })
} else {
None
}
}
fn normalize_export_fraction(fraction: Option<&str>, database_type: Option<DatabaseType>) -> String {
match fraction {
Some(fraction) if database_type == Some(DatabaseType::Mysql) && fraction.len() > 7 => fraction[..7].to_string(),
Some(fraction) => fraction.to_string(),
None => String::new(),
}
}
fn normalize_export_timezone(zone: &str) -> String {
if zone.eq_ignore_ascii_case("z") {
"+00:00".to_string()
} else {
zone.to_string()
}
}
fn is_export_timezone_offset(value: &str) -> bool {
let bytes = value.as_bytes();
bytes.len() == 6
&& matches!(bytes[0], b'+' | b'-')
&& bytes[3] == b':'
&& bytes[1].is_ascii_digit()
&& bytes[2].is_ascii_digit()
&& bytes[4].is_ascii_digit()
&& bytes[5].is_ascii_digit()
}
fn is_mysql_bit_type(column_type: &str) -> bool {
let trimmed = column_type.trim();
let lower = trimmed.to_ascii_lowercase();
lower == "bit" || lower.starts_with("bit(") || lower.starts_with("bit ")
}
fn format_mysql_bit_literal(value: &Value) -> String {
match value {
Value::Null => "NULL".to_string(),
Value::Bool(value) => {
if *value {
"b'1'".to_string()
} else {
"b'0'".to_string()
}
}
Value::Number(value) => {
let s = value.to_string();
if s == "0" || s == "1" {
format!("b'{s}'")
} else {
s
}
}
Value::String(value) => {
let trimmed = value.trim();
if trimmed.eq_ignore_ascii_case("true") {
return "b'1'".to_string();
}
if trimmed.eq_ignore_ascii_case("false") {
return "b'0'".to_string();
}
if trimmed == "0" || trimmed == "1" {
return format!("b'{trimmed}'");
}
if !trimmed.is_empty() && trimmed.bytes().all(|byte| byte == b'0' || byte == b'1') {
return format!("b'{trimmed}'");
}
format!("b'{}'", escape_mysql_compatible_export_sql_string(value))
}
other => format_export_sql_literal(other),
}
}
fn is_mysql_binary_export_type(column_type: &str) -> bool {
let lower = column_type.trim().to_ascii_lowercase();
let base = lower.split(['(', ':', ' ', '\t', '\n']).next().unwrap_or("").trim();
matches!(base, "binary" | "varbinary" | "blob" | "tinyblob" | "mediumblob" | "longblob")
}
pub(crate) fn is_mysql_spatial_export_type(column_type: &str) -> bool {
let base = column_type
.trim()
.to_ascii_lowercase()
.split(['(', ':', ' ', '\t', '\n'])
.next()
.unwrap_or("")
.trim()
.to_string();
matches!(
base.as_str(),
"geometry"
| "point"
| "linestring"
| "polygon"
| "multipoint"
| "multilinestring"
| "multipolygon"
| "geometrycollection"
| "geomcollection"
)
}
/// Database exports encode MySQL spatial cells as `DBX_WKB:<srid>:<hex>` while
/// reading them. Keeping this marker internal lets the normal JSON row shape
/// and all non-export query paths continue to expose readable WKT values.
pub(crate) fn format_mysql_spatial_export_literal(
value: &Value,
database_type: Option<DatabaseType>,
column_type: Option<&str>,
) -> Option<String> {
if database_type != Some(DatabaseType::Mysql) || !column_type.is_some_and(is_mysql_spatial_export_type) {
return None;
}
let Value::String(value) = value else {
return value.is_null().then(|| "NULL".to_string());
};
let marker = value.strip_prefix("DBX_WKB:")?;
let (srid, hex) = marker.split_once(':')?;
if srid.is_empty()
|| !srid.as_bytes().iter().all(u8::is_ascii_digit)
|| hex.is_empty()
|| hex.len() % 2 != 0
|| !hex.as_bytes().iter().all(u8::is_ascii_hexdigit)
{
return None;
}
let wkb = decode_mysql_spatial_export_wkb(hex)?;
crate::db::wkb::decode_wkb_geometry(&wkb)?;
let srid = srid.parse::<u32>().ok()?;
Some(if srid == 0 { format!("ST_GeomFromWKB(0x{hex})") } else { format!("ST_GeomFromWKB(0x{hex}, {srid})") })
}
fn decode_mysql_spatial_export_wkb(hex: &str) -> Option<Vec<u8>> {
fn nibble(byte: u8) -> Option<u8> {
match byte {
b'0'..=b'9' => Some(byte - b'0'),
b'a'..=b'f' => Some(byte - b'a' + 10),
b'A'..=b'F' => Some(byte - b'A' + 10),
_ => None,
}
}
hex.as_bytes().chunks_exact(2).map(|pair| Some((nibble(pair[0])? << 4) | nibble(pair[1])?)).collect()
}
fn format_mysql_binary_export_literal(value: &Value) -> Option<String> {
match value {
Value::Null => Some("NULL".to_string()),
Value::String(text) => format_mysql_binary_export_literal_text(text),
_ => None,
}
}
fn format_mysql_binary_export_literal_text(text: &str) -> Option<String> {
let trimmed = text.trim();
let hex = trimmed.strip_prefix("0x").or_else(|| trimmed.strip_prefix("0X"))?;
if hex.chars().all(|ch| ch.is_ascii_hexdigit()) {
// DBX exposes MySQL binary cells as 0x-prefixed hex text. Keep it as a
// MySQL hex literal so exported INSERT statements round-trip bytes.
Some(if hex.is_empty() { "X''".to_string() } else { format!("0x{hex}") })
} else {
None
}
}
fn is_export_numeric_type(column_type: &str) -> bool {
let lower = column_type.to_ascii_lowercase();
[
"int",
"integer",
"bigint",
"smallint",
"tinyint",
"mediumint",
"serial",
"number",
"numeric",
"decimal",
"dec",
"fixed",
"float",
"double",
"real",
]
.iter()
.any(|part| lower.split(|ch: char| !ch.is_ascii_alphanumeric()).any(|token| token == *part))
}
fn format_export_numeric_literal(value: &Value) -> Option<String> {
match value {
Value::Number(number) => Some(number.to_string()),
Value::String(text) if is_export_numeric_literal(text) => Some(text.to_string()),
_ => None,
}
}
fn is_export_numeric_literal(text: &str) -> bool {
if text.trim() != text || text.is_empty() {
return false;
}
text.parse::<f64>().is_ok_and(f64::is_finite)
&& text.chars().all(|ch| ch.is_ascii_digit() || matches!(ch, '+' | '-' | '.' | 'e' | 'E'))
&& text.chars().any(|ch| ch.is_ascii_digit())
}
pub fn build_export_insert_statements(options: BuildExportInsertStatementsOptions) -> Result<Vec<String>, String> {
if options.columns.is_empty() || options.rows.is_empty() {
return Ok(Vec::new());
}
let table = export_qualified_table_name(
options.database_type,
options.schema.as_deref(),
options.table_name.as_deref(),
options.qualified_table_name.as_deref(),
options.identifier_quote.as_deref(),
)?;
let insert_columns = options
.columns
.iter()
.enumerate()
.filter_map(|(index, column)| {
let column_type = options.column_types.get(index).and_then(|value| value.as_deref());
is_export_insert_column(
options.database_type,
column,
column_type,
options.column_extras.get(index).and_then(|value| value.as_deref()),
)
.then(|| {
let sqlserver_unicode_string = options.database_type == Some(DatabaseType::SqlServer)
&& column_type.is_some_and(is_sqlserver_unicode_export_type);
(index, column, sqlserver_unicode_string)
})
})
.collect::<Vec<_>>();
if insert_columns.is_empty() {
return Ok(Vec::new());
}
let batch_size = if options.database_type.is_some_and(uses_single_row_insert_statements) {
1
} else {
let requested = options.batch_size.unwrap_or(DATABASE_EXPORT_INSERT_BATCH_SIZE).max(1);
if options.database_type == Some(DatabaseType::SqlServer) {
requested.min(1000)
} else {
requested
}
};
let columns = insert_columns
.iter()
.map(|(_, column, _)| {
crate::sql_dialect::quote_table_data_identifier(
options.database_type,
column,
options.identifier_quote.as_deref(),
)
})
.collect::<Vec<_>>()
.join(", ");
let mut statements = Vec::new();
let needs_dameng_identity_insert = options.database_type == Some(DatabaseType::Dameng)
&& insert_columns.iter().any(|(index, _, _)| {
is_identity_column_extra(options.column_extras.get(*index).and_then(|value| value.as_deref()))
});
let statement_prefix = format!("INSERT INTO {table} ({columns}) VALUES ");
let statement_overhead_bytes = export_sql_statement_bytes(options.database_type, &statement_prefix) + 1;
let target_statement_bytes = DATABASE_EXPORT_TARGET_STATEMENT_BYTES;
let separator_bytes = export_sql_statement_bytes(options.database_type, ", ");
let mut current_values = String::new();
let mut current_values_bytes = 0usize;
let mut current_row_count = 0usize;
let flush_values =
|statements: &mut Vec<String>, values: &mut String, values_bytes: &mut usize, row_count: &mut usize| {
if *row_count == 0 {
return;
}
let mut insert_sql = String::with_capacity(statement_prefix.len() + values.len() + 1);
insert_sql.push_str(&statement_prefix);
insert_sql.push_str(values);
insert_sql.push(';');
if needs_dameng_identity_insert {
statements.push(wrap_dameng_identity_insert_sql_for_table(&insert_sql, &table));
} else {
statements.push(insert_sql);
}
values.clear();
*values_bytes = 0;
*row_count = 0;
};
for row in options.rows {
let mut rendered_row = String::with_capacity(insert_columns.len().saturating_mul(16).saturating_add(2));
rendered_row.push('(');
for (column_index, (index, _, sqlserver_unicode_string)) in insert_columns.iter().enumerate() {
if column_index > 0 {
rendered_row.push_str(", ");
}
let value = row.get(*index).unwrap_or(&Value::Null);
rendered_row.push_str(&format_export_sql_literal_typed(
value,
options.database_type,
options.column_types.get(*index).and_then(|value| value.as_deref()),
*sqlserver_unicode_string,
));
}
rendered_row.push(')');
let rendered_row_bytes = export_sql_statement_bytes(options.database_type, &rendered_row);
let candidate_bytes = statement_overhead_bytes
+ current_values_bytes
+ if current_row_count == 0 { 0 } else { separator_bytes }
+ rendered_row_bytes;
if current_row_count > 0 && (current_row_count >= batch_size || candidate_bytes > target_statement_bytes) {
flush_values(&mut statements, &mut current_values, &mut current_values_bytes, &mut current_row_count);
}
if current_row_count > 0 {
current_values.push_str(", ");
current_values_bytes += separator_bytes;
}
current_values.push_str(&rendered_row);
current_values_bytes += rendered_row_bytes;
current_row_count += 1;
}
flush_values(&mut statements, &mut current_values, &mut current_values_bytes, &mut current_row_count);
Ok(statements)
}
fn export_sql_statement_bytes(database_type: Option<DatabaseType>, text: &str) -> usize {
if database_type == Some(DatabaseType::SqlServer) {
text.encode_utf16().count() * 2
} else {
text.len()
}
}
pub(crate) fn is_internal_export_column(database_type: Option<DatabaseType>, column: &str) -> bool {
// Oracle-compatible ROWID is injected only to identify editable rows. It
// is not a physical table column and must never propagate into exports.
crate::sql_dialect::uses_oracle_row_id(database_type)
&& column.eq_ignore_ascii_case(crate::sql_dialect::DBX_ROWID_COLUMN)
}
fn is_postgres_tsvector_export_column(database_type: Option<DatabaseType>, column_type: Option<&str>) -> bool {
database_type == Some(DatabaseType::Postgres)
&& column_type
.map(|column_type| {
let normalized = column_type.trim().trim_matches('"').to_ascii_lowercase();
normalized == "tsvector" || normalized.ends_with(".tsvector")
})
.unwrap_or(false)
}
fn is_export_insert_column(
database_type: Option<DatabaseType>,
column: &str,
column_type: Option<&str>,
column_extra: Option<&str>,
) -> bool {
!is_internal_export_column(database_type, column)
&& !is_postgres_tsvector_export_column(database_type, column_type)
&& (database_type != Some(DatabaseType::Mysql) || !is_mysql_generated_column_extra(column_extra))
}
fn is_postgres_json_export_column(database_type: Option<DatabaseType>, column_type: Option<&str>) -> bool {
database_type == Some(DatabaseType::Postgres)
&& column_type
.map(|column_type| {
let normalized = column_type.trim().trim_matches('"').to_ascii_lowercase();
matches!(normalized.as_str(), "json" | "jsonb")
|| normalized.ends_with(".json")
|| normalized.ends_with(".jsonb")
})
.unwrap_or(false)
}
fn is_postgres_vector_export_column(database_type: Option<DatabaseType>, column_type: Option<&str>) -> bool {
database_type == Some(DatabaseType::Postgres)
&& column_type
.map(|column_type| {
let normalized = column_type.trim().trim_matches('"').to_ascii_lowercase();
let base = normalized.split(['(', ' ', '\t', '\n']).next().unwrap_or("").trim_matches('"');
matches!(base, "vector" | "halfvec") || base.ends_with(".vector") || base.ends_with(".halfvec")
})
.unwrap_or(false)
}
pub fn build_export_sql_insert(options: BuildExportSqlInsertOptions) -> Result<String, String> {
build_export_insert_statements(options.insert).map(|statements| statements.join("\n"))
}
pub fn build_database_sql_export(options: BuildDatabaseSqlExportOptions) -> Result<String, String> {
let exported_at = options.exported_at.unwrap_or_else(|| chrono::Utc::now().to_rfc3339());
let row_limit = options.row_limit_per_table.unwrap_or(DATABASE_EXPORT_ROW_LIMIT);
let insert_batch_size = options.insert_batch_size.unwrap_or(DATABASE_EXPORT_INSERT_BATCH_SIZE);
let mut lines = vec![
"-- DBX database export".to_string(),
format!("-- Database: {}", options.database_name),
format!("-- Exported at: {exported_at}"),
format!("-- Row limit per table: {row_limit}"),
String::new(),
];
for table in options.tables {
if let Some(ddl) = table.ddl.as_ref().map(|ddl| ddl.trim()).filter(|ddl| !ddl.is_empty()) {
let ddl = format_export_table_ddl(
ddl,
table.database_type,
DdlNormalizeOptions { omit_auto_increment: options.omit_auto_increment },
);
lines.push(format!("-- Structure for {}", table.display_name));
lines.push(ddl);
lines.push(String::new());
}
lines.push(format!("-- Data for {}", table.display_name));
if table.truncated {
lines.push(format!("-- Exported rows: {} (truncated at {row_limit})", table.rows.len()));
} else {
lines.push(format!("-- Exported rows: {}", table.rows.len()));
}
let inserts = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: table.database_type,
identifier_quote: table.identifier_quote.clone(),
schema: table.schema,
table_name: table.table_name,
qualified_table_name: table.qualified_table_name,
columns: table.columns,
column_types: table.column_types,
column_extras: table.column_extras,
rows: table.rows,
batch_size: Some(insert_batch_size),
})?;
if inserts.is_empty() {
lines.push("-- No rows".to_string());
} else {
lines.extend(inserts);
}
lines.push(String::new());
}
Ok(lines.join("\n"))
}
fn export_qualified_table_name(
database_type: Option<DatabaseType>,
schema: Option<&str>,
table_name: Option<&str>,
qualified_name: Option<&str>,
identifier_quote: Option<&str>,
) -> Result<String, String> {
if let Some(name) = qualified_name.filter(|name| !name.trim().is_empty()) {
return Ok(name.to_string());
}
let table_name = table_name
.filter(|name| !name.trim().is_empty())
.ok_or_else(|| "tableName is required when qualifiedTableName is not provided".to_string())?;
if crate::sql_dialect::uses_connection_identifier_quote(database_type, identifier_quote) {
return Ok(crate::sql_dialect::table_data_qualified_table_name(
database_type,
schema,
table_name,
identifier_quote,
));
}
Ok(qualified_table_name(database_type, schema, table_name))
}
fn normalize_export_table_ddl(
ddl: &str,
database_type: Option<DatabaseType>,
opts: crate::mysql_ddl_normalize::DdlNormalizeOptions,
) -> String {
if database_type != Some(DatabaseType::Mysql) {
return ddl.to_string();
}
crate::mysql_ddl_normalize::normalize_mysql_export_ddl(ddl, opts)
}
fn format_export_table_ddl(ddl: &str, database_type: Option<DatabaseType>, opts: DdlNormalizeOptions) -> String {
let ddl = normalize_export_table_ddl(ddl, database_type, opts);
let ddl = ddl.trim().trim_end_matches(';').trim_end();
format!("{ddl};")
}
fn split_postgres_export_table_triggers(ddl: &str, database_type: DatabaseType) -> (String, Vec<String>) {
if database_type != DatabaseType::Postgres {
return (ddl.to_string(), Vec::new());
}
let mut table_statements = Vec::new();
let mut trigger_statements = Vec::new();
for range in crate::db::ddl_scan::top_level_statement_ranges(ddl) {
let statement = ddl[range].trim();
if statement.is_empty() {
continue;
}
let mut words = statement.split_ascii_whitespace().take(3).map(str::to_ascii_uppercase);
let first = words.next();
let second = words.next();
let third = words.next();
if first.as_deref() == Some("CREATE")
&& (second.as_deref() == Some("TRIGGER")
|| (second.as_deref() == Some("CONSTRAINT") && third.as_deref() == Some("TRIGGER")))
{
trigger_statements.push(statement.to_string());
} else {
table_statements.push(statement.to_string());
}
}
(table_statements.join("\n"), trigger_statements)
}
fn postgres_sequence_qualified_name(schema: &str, sequence_name: &str) -> String {
let db_type = DatabaseType::Postgres;
if schema.trim().is_empty() {
quote_identifier(sequence_name, &db_type)
} else {
format!("{}.{}", quote_identifier(schema, &db_type), quote_identifier(sequence_name, &db_type))
}
}
fn generate_postgres_sequence_create_ddl(sequence: &PostgresExportSequence, schema: &str) -> String {
let qualified_name = postgres_sequence_qualified_name(schema, &sequence.name);
let cycle = if sequence.cycle { "CYCLE" } else { "NO CYCLE" };
format!(
"CREATE SEQUENCE IF NOT EXISTS {qualified_name}\n AS {data_type}\n START WITH {start_value}\n INCREMENT BY {increment}\n MINVALUE {min_value}\n MAXVALUE {max_value}\n CACHE {cache_value}\n {cycle}",
data_type = sequence.data_type,
start_value = sequence.start_value,
increment = sequence.increment,
min_value = sequence.min_value,
max_value = sequence.max_value,
cache_value = sequence.cache_value,
)
}
fn generate_postgres_sequence_owner_ddl(sequence: &PostgresExportSequence, schema: &str) -> Option<String> {
let owner_table = sequence.owner_table.as_deref()?;
let owner_column = sequence.owner_column.as_deref()?;
Some(format!(
"ALTER SEQUENCE {} OWNED BY {}.{}",
postgres_sequence_qualified_name(schema, &sequence.name),
crate::transfer::qualified_table(owner_table, schema, &DatabaseType::Postgres, None),
quote_identifier(owner_column, &DatabaseType::Postgres)
))
}
fn generate_postgres_sequence_setval_sql(sequence: &PostgresExportSequence, schema: &str) -> Option<String> {
let last_value = sequence.last_value.as_deref()?.trim();
if last_value.is_empty() {
return None;
}
let sequence_literal = quote_postgres_string_literal(&postgres_sequence_qualified_name(schema, &sequence.name));
match (sequence.owner_table.as_deref(), sequence.owner_column.as_deref()) {
(Some(owner_table), Some(owner_column)) => {
let owner_table = crate::transfer::qualified_table(owner_table, schema, &DatabaseType::Postgres, None);
let owner_column = quote_identifier(owner_column, &DatabaseType::Postgres);
Some(format!(
"SELECT setval({sequence_literal}, GREATEST(COALESCE(MAX({owner_column}), {last_value}), {last_value}), true) FROM {owner_table}"
))
}
_ => Some(format!("SELECT setval({sequence_literal}, {last_value}, true)")),
}
}
fn generate_postgres_extension_ddl(extension: &PostgresExportExtension) -> String {
// Match pg_dump: omit VERSION so the target installation selects its
// default compatible version, while preserving the source schema.
format!(
"CREATE EXTENSION IF NOT EXISTS {} WITH SCHEMA {};",
quote_identifier(&extension.name, &DatabaseType::Postgres),
quote_identifier(&extension.schema, &DatabaseType::Postgres)
)
}
async fn list_postgres_extension_members(
state: &crate::connection::AppState,
pool_key: &str,
schema: &str,
) -> Result<PostgresExtensionMembers, String> {
let pool = {
let connections = state.connections.read().await;
match connections.get(pool_key) {
Some(crate::connection::PoolKind::Postgres(pool)) => pool.clone(),
_ => return Ok(PostgresExtensionMembers::default()),
}
};
let mut members = PostgresExtensionMembers::default();
for (kind, name, signature) in crate::db::postgres::list_extension_member_objects(&pool, schema).await? {
if kind == "RELATION" {
members.relation_names.insert(name);
} else if kind == "FUNCTION" {
members.function_keys.insert((name, signature));
}
}
Ok(members)
}
fn is_postgres_extension_member_routine(object: &crate::types::ObjectInfo, members: &PostgresExtensionMembers) -> bool {
members.function_keys.contains(&(object.name.clone(), object.signature.clone().unwrap_or_default()))
}
async fn list_postgres_export_sequences(
state: &crate::connection::AppState,
pool_key: &str,
schema: &str,
selected_tables: &[String],
excluded_tables: &[String],
include_objects: bool,
fail_on_error: bool,
) -> Result<Vec<PostgresExportSequence>, String> {
let pool = {
let connections = state.connections.read().await;
match connections.get(pool_key) {
Some(crate::connection::PoolKind::Postgres(pool)) => pool.clone(),
_ => return Ok(Vec::new()),
}
};
let client = pool.get().await.map_err(|e| e.to_string())?;
let rows = client
.query(
"SELECT c.relname, \
COALESCE(format_type(s.seqtypid, NULL), 'bigint'), \
COALESCE(s.seqstart::text, '1'), \
COALESCE(s.seqmin::text, '1'), \
COALESCE(s.seqmax::text, '9223372036854775807'), \
COALESCE(s.seqincrement::text, '1'), \
COALESCE(s.seqcycle, false), \
COALESCE(s.seqcache::text, '1'), \
t.relname, \
a.attname \
FROM pg_class c \
JOIN pg_namespace n ON n.oid = c.relnamespace \
LEFT JOIN pg_sequence s ON s.seqrelid = c.oid \
LEFT JOIN pg_depend d ON d.classid = 'pg_class'::regclass \
AND d.objid = c.oid \
AND d.refclassid = 'pg_class'::regclass \
AND d.deptype IN ('a', 'i') \
LEFT JOIN pg_class t ON t.oid = d.refobjid \
LEFT JOIN pg_namespace tn ON tn.oid = t.relnamespace AND tn.nspname = n.nspname \
LEFT JOIN pg_attribute a ON a.attrelid = t.oid AND a.attnum = d.refobjsubid \
WHERE c.relkind = 'S' AND n.nspname = $1 \
ORDER BY c.relname",
&[&schema],
)
.await
.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 {
name: row.get::<_, String>(0),
data_type: row.get::<_, String>(1),
start_value: row.get::<_, String>(2),
min_value: row.get::<_, String>(3),
max_value: row.get::<_, String>(4),
increment: row.get::<_, String>(5),
cycle: row.get::<_, bool>(6),
cache_value: row.get::<_, String>(7),
last_value: None,
owner_table: row.get::<_, Option<String>>(8),
owner_column: row.get::<_, Option<String>>(9),
})
.filter(|sequence| {
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<_>>();
if sequences.is_empty() {
return Ok(sequences);
}
let last_values = client
.query(
"SELECT c.relname, pg_sequence_last_value(c.oid)::text \
FROM pg_class c \
JOIN pg_namespace n ON n.oid = c.relnamespace \
WHERE c.relkind = 'S' AND n.nspname = $1",
&[&schema],
)
.await;
match last_values {
Ok(rows) => {
for row in rows {
let name: String = row.get(0);
let last_value: Option<String> = row.get(1);
if let Some(sequence) = sequences.iter_mut().find(|sequence| sequence.name == name) {
sequence.last_value = last_value;
}
}
}
Err(error) if fail_on_error => return Err(error.to_string()),
Err(_) => {}
}
Ok(sequences)
}
pub fn is_export_cancelled_now(export_id: &str) -> bool {
EXPORT_CANCELLED.read().map(|cancelled| cancelled.contains(export_id)).unwrap_or(false)
}
pub async fn is_export_cancelled(export_id: &str) -> bool {
is_export_cancelled_now(export_id)
}
pub async fn set_export_cancelled(export_id: &str) {
if let Ok(mut cancelled) = EXPORT_CANCELLED.write() {
cancelled.insert(export_id.to_string());
}
}
pub async fn clear_export_cancelled(export_id: &str) {
if let Ok(mut cancelled) = EXPORT_CANCELLED.write() {
cancelled.remove(export_id);
}
}
pub async fn begin_database_backup_snapshot_core(
state: &crate::connection::AppState,
connection_id: &str,
database: &str,
) -> Result<DatabaseBackupSnapshot, String> {
let db_type = state
.configs
.read()
.await
.get(connection_id)
.map(|config| config.db_type)
.ok_or_else(|| format!("Connection config not found: {connection_id}"))?;
if !matches!(db_type, DatabaseType::Mysql | DatabaseType::Postgres) {
return Err("Consistent database backup snapshots are only supported for MySQL and PostgreSQL".to_string());
}
let session_id = crate::query::begin_database_backup_snapshot(state, connection_id, database).await?;
let schemas = if matches!(db_type, DatabaseType::Postgres) {
const POSTGRES_BACKUP_SCHEMAS_SQL: &str = "SELECT n.nspname FROM pg_catalog.pg_namespace n \
WHERE n.nspname NOT IN ('information_schema', 'pg_catalog', 'pg_toast') \
AND n.nspname NOT LIKE 'pg_toast_temp_%' \
AND n.nspname NOT LIKE 'pg_temp_%' ORDER BY n.nspname";
let results = crate::query::execute_in_manual_transaction(
state,
&session_id,
POSTGRES_BACKUP_SCHEMAS_SQL,
database,
None,
Some(10_000),
)
.await?;
results
.into_iter()
.next()
.map(|result| {
result
.rows
.into_iter()
.filter_map(|row| row.into_iter().next())
.filter_map(|value| value.as_str().map(str::to_string))
.collect()
})
.unwrap_or_default()
} else {
vec![database.to_string()]
};
if schemas.is_empty() {
let _ = crate::query::rollback_manual_transaction(state, &session_id).await;
return Err(format!("No schemas are available in database {database}"));
}
Ok(DatabaseBackupSnapshot { session_id, schemas })
}
/// 只有确认底层支持并发请求的多连接池才允许并发预取导出元数据。
/// SqlServer(Arc<Mutex> 串行客户端)、Agent/ExternalDriver(插件请求超时覆盖
/// 排队时间且超时会终止 sidecar)、SQLite/DuckDB 等单连接类型都必须回退串行。
fn concurrent_metadata_prefetch_allowed(pool_kind: Option<&crate::connection::PoolKind>) -> bool {
matches!(
pool_kind,
Some(crate::connection::PoolKind::Postgres(_))
| Some(crate::connection::PoolKind::Mysql(..))
| Some(crate::connection::PoolKind::ClickHouse(_))
)
}
fn database_export_metadata_prefetch_concurrency(db_type: DatabaseType) -> usize {
// PostgreSQL exports can hold a snapshot connection while metadata and UI
// requests share the base pool. Keep enough capacity available for normal
// browsing instead of filling nearly the entire ten-connection pool.
if db_type == DatabaseType::Postgres {
4
} else {
8
}
}
fn record_export_error<W: Write>(file: &mut W, fail_on_error: bool, message: String) -> Result<(), String> {
if fail_on_error {
Err(message)
} else {
writeln!(file, "-- ERROR {message}").map_err(|error| format!("Failed to write file: {error}"))
}
}
fn mysql_spatial_export_marker_expression(column: &str) -> String {
let quoted = quote_identifier(column, &DatabaseType::Mysql);
format!(
"CASE WHEN {quoted} IS NULL THEN NULL ELSE CONCAT('DBX_WKB:', ST_SRID({quoted}), ':', HEX(ST_AsWKB({quoted}))) END AS {quoted}"
)
}
fn database_export_select_list(columns: &[String], column_types: &[Option<String>], db_type: &DatabaseType) -> String {
columns
.iter()
.enumerate()
.map(|(index, column)| {
if *db_type == DatabaseType::Mysql
&& column_types.get(index).and_then(|value| value.as_deref()).is_some_and(is_mysql_spatial_export_type)
{
mysql_spatial_export_marker_expression(column)
} else {
quote_identifier(column, db_type)
}
})
.collect::<Vec<_>>()
.join(", ")
}
pub(crate) fn replace_database_export_select_list(
sql: String,
columns: &[String],
column_types: &[Option<String>],
db_type: &DatabaseType,
) -> String {
let original = columns.iter().map(|column| quote_identifier(column, db_type)).collect::<Vec<_>>().join(", ");
let replacement = database_export_select_list(columns, column_types, db_type);
if replacement == original {
return sql;
}
let prefix = format!("SELECT {original}");
if !sql.starts_with(&prefix) {
log::warn!(
"MySQL spatial database export could not replace its SELECT list; geometry columns will be exported as WKT"
);
return sql;
}
format!("SELECT {replacement}{}", &sql[prefix.len()..])
}
fn database_export_select_sql(
columns: &[String],
column_types: &[Option<String>],
table: &str,
schema: &str,
db_type: &DatabaseType,
) -> String {
let columns = database_export_select_list(columns, column_types, db_type);
let table = crate::transfer::qualified_table(table, schema, db_type, None);
format!("SELECT {columns} FROM {table}")
}
fn write_database_export_rows<W: Write>(
file: &mut W,
rows: &[Vec<Value>],
columns: &[String],
column_types: &[Option<String>],
column_extras: &[Option<String>],
table: &str,
schema: &str,
db_type: &DatabaseType,
) -> Result<(), String> {
let insert_indices = columns
.iter()
.enumerate()
.filter(|(index, column)| {
is_export_insert_column(
Some(*db_type),
column,
column_types.get(*index).and_then(|value| value.as_deref()),
column_extras.get(*index).and_then(|value| value.as_deref()),
)
})
.map(|(index, _)| index)
.collect::<Vec<_>>();
if insert_indices.is_empty() {
return Ok(());
}
let filtered_columns;
let filtered_column_types;
let filtered_column_extras;
let filtered_rows;
let (insert_columns, insert_column_types, insert_column_extras, insert_rows) = if insert_indices.len()
== columns.len()
{
(columns, column_types, column_extras, rows)
} else {
filtered_columns = insert_indices.iter().map(|index| columns[*index].clone()).collect::<Vec<_>>();
filtered_column_types =
insert_indices.iter().map(|index| column_types.get(*index).cloned().unwrap_or(None)).collect::<Vec<_>>();
filtered_column_extras =
insert_indices.iter().map(|index| column_extras.get(*index).cloned().unwrap_or(None)).collect::<Vec<_>>();
filtered_rows = rows
.iter()
.map(|row| insert_indices.iter().map(|index| row.get(*index).cloned().unwrap_or(Value::Null)).collect())
.collect::<Vec<Vec<Value>>>();
(
filtered_columns.as_slice(),
filtered_column_types.as_slice(),
filtered_column_extras.as_slice(),
filtered_rows.as_slice(),
)
};
// Database exports use the same table qualification as the SELECT/DDL
// path and the legacy typed INSERT writer. In particular, MySQL uses the
// selected database rather than a schema-qualified table name.
let qualified_table_name = crate::transfer::qualified_table(table, schema, db_type, None);
// Batch database exports do not currently thread a per-connection identifier
// quote through BuildDatabaseSqlExportOptions; Kingbase MySQL-compat users
// should fall back to the single-table export path which carries the quote.
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(*db_type),
identifier_quote: None,
schema: (!schema.is_empty()).then(|| schema.to_string()),
table_name: Some(table.to_string()),
qualified_table_name: Some(qualified_table_name),
columns: insert_columns.to_vec(),
column_types: insert_column_types.to_vec(),
column_extras: insert_column_extras.to_vec(),
rows: insert_rows.to_vec(),
batch_size: Some(DATABASE_EXPORT_INSERT_BATCH_SIZE),
})?;
for statement in statements {
writeln!(file, "{statement}\n").map_err(|error| format!("Failed to write file: {error}"))?;
}
Ok(())
}
fn emit_database_export_running(
on_progress: &(impl Fn(ExportProgress) + Sync),
export_id: &str,
current_object: impl Into<String>,
object_index: usize,
total_objects: usize,
rows_exported: u64,
preparing: bool,
) {
on_progress(ExportProgress {
export_id: export_id.to_string(),
current_object: current_object.into(),
object_index,
total_objects,
rows_exported,
total_rows: None,
status: ExportStatus::Running,
error: None,
preparing,
});
}
fn export_destination_state_key(dir: &std::path::Path) -> String {
format!("database_export_destination:{}", dir.to_string_lossy())
}
/// Records the destination identity for a scheduled backup as soon as it is
/// configured, not just after its first successful export. Scheduled plans
/// live in the frontend and may not run for hours after being saved; without
/// an eager record here, a destination whose mount disappears before its
/// very first run is indistinguishable from a brand-new local folder to
/// `ensure_export_destination_dir` (both have no recorded state) and gets
/// silently recreated on the wrong filesystem. The schedule editor only
/// accepts directories selected from the filesystem, so a missing path here
/// means the destination vanished before its identity could be recorded and
/// the schedule must not be saved. See #6327.
pub async fn record_export_destination_identity(
state: &crate::connection::AppState,
dir: &std::path::Path,
) -> Result<(), String> {
if !dir.is_dir() {
return Err(format!(
"Backup directory {} does not exist or is not a directory. Select an existing destination before saving the schedule.",
dir.display()
));
}
save_export_destination_dir_device_id(state, dir).await
}
async fn save_export_destination_dir_device_id(
state: &crate::connection::AppState,
dir: &std::path::Path,
) -> Result<(), String> {
let value = match export_destination_device_id_for_path(dir) {
Some(dev) => dev.to_le_bytes().to_vec(),
None => Vec::new(),
};
state.storage.save_state(&export_destination_state_key(dir), &value, "application/octet-stream").await
}
/// Ensures `dir` exists for an export destination, without ever silently
/// recreating a directory that previously produced a successful export (or
/// was recorded via [`record_export_destination_identity`]) and has since
/// disappeared. Auto-creating on every run is what the original fix for
/// #6109 did, but that is unsafe for a destination on a removable or network
/// drive: if the mount is temporarily gone when a run executes, blindly
/// recreating the path resurrects it on the local root filesystem and the
/// export "succeeds" while silently writing to the wrong disk. A directory
/// dbx has never seen before is safe to create (normal first-time
/// configuration of a local folder); a directory dbx has seen before but
/// that is now missing, or that now resolves to a different filesystem than
/// last time, is refused instead. See #6327.
///
/// Returns the device identity that was just verified (or recorded for a
/// newly created directory), if the current platform can determine one, so
/// the caller can re-verify it against the file it actually opens -- the
/// directory check here and the later `File::create` are separate
/// operations, and the mount can change in between.
async fn ensure_export_destination_dir(
state: &crate::connection::AppState,
dir: &std::path::Path,
) -> Result<Option<u64>, String> {
let key = export_destination_state_key(dir);
let recorded_dev: Option<Option<u64>> = state
.storage
.load_state(&key)
.await?
.map(|(bytes, _content_type)| <[u8; 8]>::try_from(bytes.as_slice()).ok().map(u64::from_le_bytes));
if dir.is_dir() {
if let Some(Some(recorded_dev)) = recorded_dev {
if let Some(current_dev) = export_destination_device_id_for_path(dir) {
if current_dev != recorded_dev {
return Err(format!(
"Backup directory {} now resolves to a different filesystem than the last \
successful export to this location. Refusing to write here automatically -- \
if this directory is on a removable or network drive, make sure the correct \
drive is connected before running the backup.",
dir.display()
));
}
}
}
} else {
if recorded_dev.is_some() {
return Err(format!(
"Backup directory {} is missing. It was configured or previously used for exports to \
this location, so dbx will not recreate it automatically -- if this is on a removable \
or network drive, reconnect it and try again.",
dir.display()
));
}
std::fs::create_dir_all(dir).map_err(|e| format!("Failed to create backup directory: {e}"))?;
}
save_export_destination_dir_device_id(state, dir).await?;
Ok(export_destination_device_id_for_path(dir))
}
/// Whether a destination's identity, checked once via [`ensure_export_destination_dir`]
/// and then again against the file dbx actually opened, indicates the mount
/// changed in between. An unknown expected identity has nothing to compare,
/// but once an expected identity is known, failing to identify the opened
/// handle must fail closed rather than allowing an unverified write.
fn export_destination_identity_mismatch(expected: Option<u64>, actual: Option<u64>) -> bool {
expected.is_some_and(|expected| actual != Some(expected))
}
#[cfg(unix)]
fn export_destination_device_id_for_path(dir: &std::path::Path) -> Option<u64> {
use std::os::unix::fs::MetadataExt;
std::fs::metadata(dir).ok().map(|metadata| metadata.dev())
}
#[cfg(unix)]
fn export_destination_device_id_for_file(file: &std::fs::File) -> Option<u64> {
use std::os::unix::fs::MetadataExt;
file.metadata().ok().map(|metadata| metadata.dev())
}
#[cfg(windows)]
fn export_destination_device_id_for_path(dir: &std::path::Path) -> Option<u64> {
windows_export_destination::device_id_for_path(dir)
}
#[cfg(windows)]
fn export_destination_device_id_for_file(file: &std::fs::File) -> Option<u64> {
windows_export_destination::device_id_for_handle(file)
}
#[cfg(not(any(unix, windows)))]
fn export_destination_device_id_for_path(_dir: &std::path::Path) -> Option<u64> {
None
}
#[cfg(not(any(unix, windows)))]
fn export_destination_device_id_for_file(_file: &std::fs::File) -> Option<u64> {
None
}
/// Windows has no stable `std` API for a directory or file's volume identity
/// (`MetadataExt::volume_serial_number` is still gated behind the unstable
/// `windows_by_handle` feature), so this queries `dwVolumeSerialNumber` from
/// `BY_HANDLE_FILE_INFORMATION` directly. Using the *handle* rather than
/// re-resolving the path is what makes `export_destination_device_id_for_file`
/// safe to call on an already-open `File`: it reports the volume the handle
/// was actually opened against, not whatever currently sits at that path.
#[cfg(windows)]
mod windows_export_destination {
use std::os::windows::ffi::OsStrExt;
use std::os::windows::io::AsRawHandle;
use std::path::Path;
use windows_sys::Win32::Foundation::{CloseHandle, HANDLE, INVALID_HANDLE_VALUE};
use windows_sys::Win32::Storage::FileSystem::{
CreateFileW, GetFileInformationByHandle, BY_HANDLE_FILE_INFORMATION, FILE_FLAG_BACKUP_SEMANTICS,
FILE_SHARE_DELETE, FILE_SHARE_READ, FILE_SHARE_WRITE, OPEN_EXISTING,
};
pub(super) fn device_id_for_path(path: &Path) -> Option<u64> {
let mut wide: Vec<u16> = path.as_os_str().encode_wide().collect();
wide.push(0);
let handle = unsafe {
CreateFileW(
wide.as_ptr(),
0,
FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
std::ptr::null(),
OPEN_EXISTING,
FILE_FLAG_BACKUP_SEMANTICS,
std::ptr::null_mut(),
)
};
if handle == INVALID_HANDLE_VALUE {
return None;
}
let result = volume_serial_number(handle);
unsafe { CloseHandle(handle) };
result
}
pub(super) fn device_id_for_handle(file: &std::fs::File) -> Option<u64> {
volume_serial_number(file.as_raw_handle() as HANDLE)
}
fn volume_serial_number(handle: HANDLE) -> Option<u64> {
let mut info: BY_HANDLE_FILE_INFORMATION = unsafe { std::mem::zeroed() };
let ok = unsafe { GetFileInformationByHandle(handle, &mut info) };
(ok != 0).then_some(info.dwVolumeSerialNumber as u64)
}
}
pub async fn export_database_sql_core(
state: &crate::connection::AppState,
request: &DatabaseExportRequest,
on_progress: impl Fn(ExportProgress) + Sync,
) -> Result<(), String> {
// Keep the large export state machine on the heap. Besides making the
// caller future small, this prevents the metadata-prefetch locals from
// exhausting the bounded stack used by test and runtime worker threads.
Box::pin(export_database_sql_core_inner(state, request, on_progress)).await
}
async fn export_database_sql_core_inner(
state: &crate::connection::AppState,
request: &DatabaseExportRequest,
on_progress: impl Fn(ExportProgress) + Sync,
) -> Result<(), String> {
let _snapshot_keep_alive = if let Some(snapshot_session_id) = request.snapshot_session_id.as_deref() {
Some(crate::query::keep_manual_transaction_alive(state, snapshot_session_id).await?)
} else {
None
};
// Emit immediately so the UI is never blank while we list schema metadata.
emit_database_export_running(&on_progress, &request.export_id, "", 0, 0, 0, true);
// 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))?;
// 2. Get pool
let client_session_id = database_export_client_session_id(&request.export_id);
let pool_key = Box::pin(state.get_or_create_pool_for_session(
&request.connection_id,
Some(&request.database),
Some(&client_session_id),
))
.await?;
// 3. List tables
let all_tables = Box::pin(crate::schema::list_tables_core(
state,
&request.connection_id,
&request.database,
&request.schema,
None,
None,
None,
None,
None,
))
.await?;
// 4. Create file
let mut expected_destination_dev = None;
if let Some(parent) = std::path::Path::new(&request.file_path).parent() {
if !parent.as_os_str().is_empty() {
expected_destination_dev = ensure_export_destination_dir(state, parent).await?;
}
}
let file = std::fs::File::create(&request.file_path).map_err(|e| format!("Failed to write file: {e}"))?;
// The directory check above and this `File::create` are separate
// operations: the mount can disappear and be replaced by something else
// at the same path in between. Re-check the identity of the handle we
// actually opened, not just the path, and refuse to keep a backup that
// landed on the wrong filesystem. See #6327.
if export_destination_identity_mismatch(expected_destination_dev, export_destination_device_id_for_file(&file)) {
drop(file);
let _ = std::fs::remove_file(&request.file_path);
return Err(format!(
"Backup destination for {} changed while opening the output file -- the directory now \
resolves to a different filesystem than the one just verified. If a removable or network \
drive was disconnected and reconnected, retry the backup.",
request.file_path
));
}
let mut file = BufWriter::new(file);
let create_database_preamble = if request.include_create_database && matches!(db_type, DatabaseType::Mysql) {
Some(mysql_database_export_preamble_for_request(state, request).await)
} else {
None
};
// 5. Write header
let timestamp = chrono::Local::now().format("%Y-%m-%d %H:%M:%S");
writeln!(file, "-- Database export: {}", request.database).map_err(|e| format!("Failed to write file: {e}"))?;
writeln!(file, "-- Date: {timestamp}").map_err(|e| format!("Failed to write file: {e}"))?;
writeln!(file, "-- Generated by DBX").map_err(|e| format!("Failed to write file: {e}"))?;
writeln!(file).map_err(|e| format!("Failed to write file: {e}"))?;
if let Some(preamble) = create_database_preamble {
writeln!(file, "-- Database setup").map_err(|e| format!("Failed to write file: {e}"))?;
writeln!(file, "{preamble}").map_err(|e| format!("Failed to write file: {e}"))?;
}
// 6. For MySQL: disable foreign key checks
if matches!(db_type, DatabaseType::Mysql) {
writeln!(file, "SET FOREIGN_KEY_CHECKS = 0;\n").map_err(|e| format!("Failed to write file: {e}"))?;
}
// 7. Separate tables and views
let postgres_extension_members =
if matches!(db_type, DatabaseType::Postgres) && (request.include_structure || request.include_objects) {
match list_postgres_extension_members(state, &pool_key, &request.schema).await {
Ok(members) => members,
Err(e) => {
record_export_error(&mut file, request.fail_on_error, format!("reading extension members: {e}"))?;
PostgresExtensionMembers::default()
}
}
} else {
PostgresExtensionMembers::default()
};
let postgres_extensions = if request.include_structure && matches!(db_type, DatabaseType::Postgres) {
match crate::schema::list_extensions_core(
state,
&request.connection_id,
&request.database,
Some(&request.schema),
)
.await
{
Ok(extensions) => extensions
.into_iter()
.map(|extension| PostgresExportExtension {
name: extension.name,
schema: extension.schema.unwrap_or_else(|| request.schema.clone()),
})
.collect(),
Err(e) => {
record_export_error(&mut file, request.fail_on_error, format!("exporting extensions: {e}"))?;
Vec::new()
}
}
} else {
Vec::new()
};
let all_tables = filter_export_table_infos(all_tables, &request.selected_tables, &request.excluded_tables)
.into_iter()
.filter(|table| !postgres_extension_members.relation_names.contains(&table.name))
.collect::<Vec<_>>();
let mut tables: Vec<_> = all_tables.iter().filter(|t| !t.table_type.contains("VIEW")).collect();
let mut views: Vec<_> = all_tables.iter().filter(|t| t.table_type.contains("VIEW")).collect();
if request.include_objects && db_type == DatabaseType::Mysql && views.len() > 1 {
match list_mysql_export_view_dependencies(state, &request.connection_id, &request.database).await {
Ok(dependencies) => views = sort_export_views_by_dependencies(&views, &dependencies),
Err(error) => {
log::debug!(
"[database-export] failed to resolve MySQL view dependencies for {}: {error}",
request.database
);
}
}
}
let postgres_sequences = if request.include_structure && matches!(db_type, DatabaseType::Postgres) {
match list_postgres_export_sequences(
state,
&pool_key,
&request.schema,
&request.selected_tables,
&request.excluded_tables,
request.include_objects,
request.fail_on_error,
)
.await
{
Ok(sequences) => sequences,
Err(e) => {
record_export_error(&mut file, request.fail_on_error, format!("exporting sequences: {e}"))?;
Vec::new()
}
}
} else {
Vec::new()
};
// Sort tables by foreign key dependency so referenced (parent) tables are
// exported before referencing (child) tables. MySQL exports always emit
// `SET FOREIGN_KEY_CHECKS = 0`, so both the generated DDL and any INSERTs
// already tolerate child-before-parent ordering; reordering away from the
// alphabetical listing there only makes the output look shuffled to users
// without providing any correctness benefit.
if tables.len() > 1 && db_type != DatabaseType::Mysql {
let table_names: Vec<String> = tables.iter().map(|t| t.name.clone()).collect();
match crate::transfer::sort_tables_by_fk_dependency(
state,
&request.connection_id,
&request.database,
&request.schema,
&table_names,
true,
)
.await
{
Ok(sorted_names) => {
tables.sort_by_key(|t| sorted_names.iter().position(|n| n == &t.name).unwrap_or(usize::MAX));
}
Err(error) if request.fail_on_error => {
return Err(format!("Failed to resolve table dependency order: {error}"));
}
Err(_) => {}
}
}
// 8. Discover optional schema-wide objects before calculating workload.
let mut procedures: Vec<crate::types::ObjectInfo> = Vec::new();
let mut functions: Vec<crate::types::ObjectInfo> = Vec::new();
if exports_database_routines(request) {
match crate::schema::list_objects_core(
state,
&request.connection_id,
&request.database,
&request.schema,
None,
None,
None,
None,
)
.await
{
Ok(objects) => {
for obj in &objects {
let ot = obj.object_type.to_uppercase();
if is_postgres_extension_member_routine(obj, &postgres_extension_members) {
continue;
}
if ot.contains("PROCEDURE") {
procedures.push(obj.clone());
} else if ot.contains("FUNCTION") {
functions.push(obj.clone());
}
}
}
Err(error) if request.fail_on_error => return Err(format!("Failed to list database objects: {error}")),
Err(_) => {}
}
}
// A determinate total must describe the same object categories guarded by
// the execution branches below, not every object discovered in the schema.
let total_objects = database_export_total_objects(
request,
&DatabaseExportObjectCounts {
tables: tables.len(),
views: views.len(),
sequences: postgres_sequences.len(),
extensions: postgres_extensions.len(),
procedures: procedures.len(),
functions: functions.len(),
},
);
let mut object_index: usize = 0;
let mut total_rows_exported = 0_u64;
let mut deferred_postgres_triggers = Vec::new();
// total_objects is known later for the write phase; preparing updates stay
// presence-only so the UI does not show a counter that later resets.
emit_database_export_running(&on_progress, &request.export_id, "", 0, 0, 0, true);
let batch_size = if request.batch_size == 0 { 1000 } else { request.batch_size };
// 预取各表的 DDL 与列元数据:逐表串行往返在多表数据库上是整库导出耗时的
// 主要来源(每表 1-2 次网络往返 × 表数)。有界并发预取后,下方写出循环仍按
// 原顺序消费,文件内容与逐表查询完全一致。放在写出循环之前,避免准备态与导出态来回跳。
struct PrefetchedTableMetadata {
ddl: Option<Result<String, String>>,
columns: Option<Result<Vec<crate::db::ColumnInfo>, String>>,
}
let mut prefetched_table_metadata: Vec<Option<PrefetchedTableMetadata>> = Vec::new();
prefetched_table_metadata.resize_with(tables.len(), || None);
// 防护门按「实际连接池种类」放行,而非数据库类型的能力标记:只有确认底层
// 支持并发请求的多连接池(Postgres/MySQL/ClickHouse)才启用并发预取。反例:
// 原生 SQL Server 是 Arc<Mutex<..>> 串行客户端;Agent/外部 JDBC(含 SQL Server
// legacy profile、PrestoSQL 等路由结果)的插件请求超时覆盖排队时间且超时会
// 终止 sidecar;SQLite/DuckDB 等为单连接。被挡住的场景预取 Vec 保持全 None,
// 写出循环内的 None 回退路径即原有的逐表串行直查行为。
let concurrent_prefetch_is_safe =
match state.get_or_create_pool(&request.connection_id, Some(&request.database)).await {
Ok(metadata_pool_key) => {
concurrent_metadata_prefetch_allowed(state.connections.read().await.get(&metadata_pool_key))
}
// 建池失败时不预取,让写出循环的直查路径按原有方式报告错误
Err(_) => false,
};
if concurrent_prefetch_is_safe
&& exports_database_tables(request)
&& !tables.is_empty()
&& !is_export_cancelled(&request.export_id).await
{
use futures::StreamExt;
let prefetch_targets: Vec<(usize, String)> =
tables.iter().enumerate().map(|(index, table_info)| (index, table_info.name.clone())).collect();
let mut prefetch_stream =
futures::stream::iter(prefetch_targets.into_iter().map(|(index, table_name)| async move {
if is_export_cancelled_now(&request.export_id) {
return (index, PrefetchedTableMetadata { ddl: None, columns: None });
}
let ddl = if request.include_structure {
Some(
crate::schema::get_table_relation_export_ddl_core(
state,
&request.connection_id,
&request.database,
&request.schema,
&table_name,
None,
)
.await,
)
} else {
None
};
if is_export_cancelled_now(&request.export_id) {
return (index, PrefetchedTableMetadata { ddl, columns: None });
}
let columns = if request.include_data {
Some(
crate::schema::get_columns_core(
state,
&request.connection_id,
&request.database,
&request.schema,
&table_name,
)
.await,
)
} else {
None
};
(index, PrefetchedTableMetadata { ddl, columns })
}))
.buffer_unordered(database_export_metadata_prefetch_concurrency(db_type));
while let Some((index, metadata)) = prefetch_stream.next().await {
prefetched_table_metadata[index] = Some(metadata);
if let Some(table_info) = tables.get(index) {
// Presence-only updates: no prepare counter that later resets to 0/N.
emit_database_export_running(
&on_progress,
&request.export_id,
table_info.name.clone(),
0,
0,
total_rows_exported,
true,
);
}
// 取消后不再调度新的预取任务(已在途的任务随 stream 释放而中止),
// 写出循环入口的取消检查负责最终收尾
if is_export_cancelled_now(&request.export_id) {
break;
}
}
}
for extension in &postgres_extensions {
if is_export_cancelled(&request.export_id).await {
return Err("Export cancelled".to_string());
}
emit_database_export_running(
&on_progress,
&request.export_id,
extension.name.clone(),
object_index,
total_objects,
total_rows_exported,
false,
);
writeln!(file, "{}\n", generate_postgres_extension_ddl(extension))
.map_err(|e| format!("Failed to write file: {e}"))?;
object_index += 1;
}
for sequence in postgres_sequences.iter().filter(|sequence| sequence.owner_table.is_none()) {
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,
);
writeln!(file, "{};\n", generate_postgres_sequence_create_ddl(sequence, &request.schema))
.map_err(|e| format!("Failed to write file: {e}"))?;
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 {
on_progress(ExportProgress {
export_id: request.export_id.clone(),
current_object: table_info.name.clone(),
object_index,
total_objects,
rows_exported: total_rows_exported,
total_rows: None,
status: ExportStatus::Cancelled,
error: None,
preparing: false,
});
return Ok(());
}
let table_name = &table_info.name;
// Emit Running progress
on_progress(ExportProgress {
export_id: request.export_id.clone(),
current_object: table_name.clone(),
object_index,
total_objects,
rows_exported: total_rows_exported,
total_rows: None,
status: ExportStatus::Running,
error: None,
preparing: false,
});
// Export structure
if request.include_structure {
if request.drop_table_if_exists {
writeln!(file, "{}\n", drop_table_if_exists_sql(table_name, &request.schema, &db_type))
.map_err(|e| format!("Failed to write file: {e}"))?;
}
for sequence in postgres_sequences
.iter()
.filter(|sequence| sequence.owner_table.as_deref() == Some(table_name.as_str()))
{
on_progress(ExportProgress {
export_id: request.export_id.clone(),
current_object: sequence.name.clone(),
object_index,
total_objects,
rows_exported: total_rows_exported,
total_rows: None,
status: ExportStatus::Running,
error: None,
preparing: false,
});
writeln!(file, "{};\n", generate_postgres_sequence_create_ddl(sequence, &request.schema))
.map_err(|e| format!("Failed to write file: {e}"))?;
object_index += 1;
}
let ddl_result = match prefetched_table_metadata
.get_mut(table_index)
.and_then(|m| m.as_mut())
.and_then(|m| m.ddl.take())
{
Some(result) => result,
None => {
crate::schema::get_table_relation_export_ddl_core(
state,
&request.connection_id,
&request.database,
&request.schema,
table_name,
None,
)
.await
}
};
match ddl_result {
Ok(ddl) => {
let (ddl, triggers) = split_postgres_export_table_triggers(&ddl, db_type);
deferred_postgres_triggers.extend(triggers);
let ddl = format_export_table_ddl(
&ddl,
Some(db_type),
DdlNormalizeOptions { omit_auto_increment: request.omit_auto_increment },
);
writeln!(file, "{ddl}\n").map_err(|e| format!("Failed to write file: {e}"))?;
}
Err(e) => {
record_export_error(
&mut file,
request.fail_on_error,
format!("exporting table structure {table_name}: {e}"),
)?;
}
}
}
// Export data
if request.include_data {
// Get columns
let columns_result = match prefetched_table_metadata
.get_mut(table_index)
.and_then(|m| m.as_mut())
.and_then(|m| m.columns.take())
{
Some(result) => result,
None => {
crate::schema::get_columns_core(
state,
&request.connection_id,
&request.database,
&request.schema,
table_name,
)
.await
}
};
let columns = match columns_result {
Ok(cols) => cols,
Err(e) => {
record_export_error(
&mut file,
request.fail_on_error,
format!("exporting columns for table {table_name}: {e}"),
)?;
object_index += 1;
continue;
}
};
let col_names = columns.iter().map(|c| c.name.clone()).collect::<Vec<_>>();
let col_types = columns.iter().map(|c| Some(c.data_type.clone())).collect::<Vec<_>>();
let col_extras = columns.iter().map(|c| c.extra.clone()).collect::<Vec<_>>();
if !col_names.is_empty() {
if let Some(snapshot_session_id) = request.snapshot_session_id.as_deref() {
let sql = database_export_select_sql(&col_names, &col_types, table_name, &request.schema, &db_type);
crate::query::stream_rows_in_manual_transaction(
state,
snapshot_session_id,
&sql,
batch_size,
|rows| {
if is_export_cancelled_now(&request.export_id) {
return Err("Export cancelled".to_string());
}
write_database_export_rows(
&mut file,
&rows,
&col_names,
&col_types,
&col_extras,
table_name,
&request.schema,
&db_type,
)?;
total_rows_exported += rows.len() as u64;
on_progress(ExportProgress {
export_id: request.export_id.clone(),
current_object: table_name.clone(),
object_index,
total_objects,
rows_exported: total_rows_exported,
total_rows: None,
status: ExportStatus::Running,
error: None,
preparing: false,
});
Ok(())
},
)
.await?;
} else {
// Exact COUNT(*) is deliberately skipped for manual database exports. It adds a full
// table scan before the real read and does not improve correctness. Progress reports
// exported rows while total_rows remains indeterminate.
let total_rows = None;
let primary_keys = columns
.iter()
.filter(|column| column.is_primary_key)
.map(|column| column.name.clone())
.collect::<Vec<_>>();
let primary_key_indices = primary_keys
.iter()
.filter_map(|primary_key| col_names.iter().position(|column| column == primary_key))
.collect::<Vec<_>>();
let use_keyset = !primary_keys.is_empty() && primary_key_indices.len() == primary_keys.len();
let mut last_primary_key_values = Vec::new();
let mut offset = 0_u64;
loop {
if is_export_cancelled(&request.export_id).await {
on_progress(ExportProgress {
export_id: request.export_id.clone(),
current_object: table_name.clone(),
object_index,
total_objects,
rows_exported: total_rows_exported,
total_rows,
status: ExportStatus::Cancelled,
error: None,
preparing: false,
});
return Ok(());
}
let sql = if use_keyset {
let sql = keyset_pagination_sql_with_identifier_quote(
&col_names,
table_name,
&request.schema,
&db_type,
&primary_keys,
&last_primary_key_values,
batch_size,
None,
);
replace_database_export_select_list(sql, &col_names, &col_types, &db_type)
} else {
let sql = crate::transfer::pagination_sql(
&col_names,
table_name,
&request.schema,
&db_type,
offset,
batch_size,
);
replace_database_export_select_list(sql, &col_names, &col_types, &db_type)
};
let result = match crate::transfer::execute_read_on_pool(state, &pool_key, &sql).await {
Ok(result) => result,
Err(error) => {
record_export_error(
&mut file,
request.fail_on_error,
format!("exporting data for table {table_name}: {error}"),
)?;
break;
}
};
let row_count = result.rows.len();
if row_count == 0 {
break;
}
write_database_export_rows(
&mut file,
&result.rows,
&col_names,
&col_types,
&col_extras,
table_name,
&request.schema,
&db_type,
)?;
total_rows_exported += row_count as u64;
if use_keyset {
if let Some(last_row) = result.rows.last() {
last_primary_key_values =
primary_key_indices.iter().map(|&index| last_row[index].clone()).collect();
}
} else {
offset += row_count as u64;
}
on_progress(ExportProgress {
export_id: request.export_id.clone(),
current_object: table_name.clone(),
object_index,
total_objects,
rows_exported: total_rows_exported,
total_rows,
status: ExportStatus::Running,
error: None,
preparing: false,
});
if row_count < batch_size {
break;
}
}
}
}
}
object_index += 1;
}
if request.include_structure && !postgres_sequences.is_empty() {
for sequence in &postgres_sequences {
if let Some(sql) = generate_postgres_sequence_owner_ddl(sequence, &request.schema) {
writeln!(file, "{};\n", sql).map_err(|e| format!("Failed to write file: {e}"))?;
}
}
for sequence in &postgres_sequences {
if let Some(sql) = generate_postgres_sequence_setval_sql(sequence, &request.schema) {
writeln!(file, "{};\n", sql).map_err(|e| format!("Failed to write file: {e}"))?;
}
}
}
// Export views (if include_objects)
if request.include_objects {
for view_info in &views {
if is_export_cancelled(&request.export_id).await {
return Err("Export cancelled".to_string());
}
let view_name = &view_info.name;
on_progress(ExportProgress {
export_id: request.export_id.clone(),
current_object: view_name.clone(),
object_index,
total_objects,
rows_exported: total_rows_exported,
total_rows: None,
status: ExportStatus::Running,
error: None,
preparing: false,
});
match crate::schema::get_object_source_core(
state,
&request.connection_id,
&request.database,
&request.schema,
view_name,
crate::db::ObjectSourceKind::View,
None,
None,
)
.await
{
Ok(obj_source) => {
let source = build_database_export_object_source_sql(
db_type,
&ObjectSourceKind::View,
view_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 view {view_name}: {e}"))?;
}
}
object_index += 1;
}
// Export procedures
for procedure in &procedures {
if is_export_cancelled(&request.export_id).await {
return Err("Export cancelled".to_string());
}
let proc_name = &procedure.name;
on_progress(ExportProgress {
export_id: request.export_id.clone(),
current_object: proc_name.clone(),
object_index,
total_objects,
rows_exported: total_rows_exported,
total_rows: None,
status: ExportStatus::Running,
error: None,
preparing: false,
});
match crate::schema::get_object_source_core(
state,
&request.connection_id,
&request.database,
&request.schema,
proc_name,
crate::db::ObjectSourceKind::Procedure,
procedure.signature.as_deref(),
None,
)
.await
{
Ok(obj_source) => {
let source = build_database_export_object_source_sql(
db_type,
&ObjectSourceKind::Procedure,
proc_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 procedure {proc_name}: {e}"),
)?;
}
}
object_index += 1;
}
// Export functions
for function in &functions {
if is_export_cancelled(&request.export_id).await {
return Err("Export cancelled".to_string());
}
let func_name = &function.name;
on_progress(ExportProgress {
export_id: request.export_id.clone(),
current_object: func_name.clone(),
object_index,
total_objects,
rows_exported: total_rows_exported,
total_rows: None,
status: ExportStatus::Running,
error: None,
preparing: false,
});
match crate::schema::get_object_source_core(
state,
&request.connection_id,
&request.database,
&request.schema,
func_name,
crate::db::ObjectSourceKind::Function,
function.signature.as_deref(),
None,
)
.await
{
Ok(obj_source) => {
let source = build_database_export_object_source_sql(
db_type,
&ObjectSourceKind::Function,
func_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 function {func_name}: {e}"),
)?;
}
}
object_index += 1;
}
}
// PostgreSQL trigger definitions reference their trigger functions. The
// table DDL builder returns both statements together, while schema-wide
// routines are exported below the tables. Keep triggers part of table
// structure, but write them only after routines so the resulting script
// is executable in file order.
for trigger in deferred_postgres_triggers {
writeln!(file, "{trigger}\n").map_err(|e| format!("Failed to write file: {e}"))?;
}
// For MySQL: re-enable foreign key checks
if matches!(db_type, DatabaseType::Mysql) {
writeln!(file, "SET FOREIGN_KEY_CHECKS = 1;").map_err(|e| format!("Failed to write file: {e}"))?;
}
file.flush().map_err(|e| format!("Failed to finalize export file: {e}"))?;
// Emit Done progress
on_progress(ExportProgress {
export_id: request.export_id.clone(),
current_object: String::new(),
object_index,
total_objects,
rows_exported: total_rows_exported,
total_rows: None,
status: ExportStatus::Done,
error: None,
preparing: false,
});
Ok(())
}
fn filter_export_table_infos(
tables: Vec<crate::types::TableInfo>,
selected_tables: &[String],
excluded_tables: &[String],
) -> Vec<crate::types::TableInfo> {
let selected: HashSet<&str> = selected_tables.iter().map(String::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 {
format!("DROP TABLE IF EXISTS {};", crate::transfer::qualified_table(table_name, schema, db_type, None))
}
fn build_database_export_object_source_sql(
database_type: DatabaseType,
object_type: &ObjectSourceKind,
object_name: &str,
source: &str,
drop_if_exists: bool,
) -> String {
let source = build_export_object_source_sql(database_type, object_type.clone(), source);
if source.is_empty() || !drop_if_exists || database_type != DatabaseType::Mysql {
return source;
}
let object_type = match object_type {
ObjectSourceKind::View => "VIEW",
ObjectSourceKind::Procedure => "PROCEDURE",
ObjectSourceKind::Function => "FUNCTION",
_ => return source,
};
let object_name = quote_identifier(object_name, &DatabaseType::Mysql);
format!("DROP {object_type} IF EXISTS {object_name};\n{source}")
}
#[cfg(test)]
mod tests {
use super::{
build_database_export_object_source_sql, build_database_sql_export, build_export_insert_statements,
database_export_select_sql, database_export_total_objects, drop_table_if_exists_sql,
ensure_export_destination_dir, export_destination_identity_mismatch, filter_export_table_infos,
format_export_sql_literal, format_export_table_ddl, format_mysql_spatial_export_literal,
generate_postgres_extension_ddl, generate_postgres_sequence_create_ddl, generate_postgres_sequence_owner_ddl,
generate_postgres_sequence_setval_sql, is_postgres_extension_member_routine, mysql_database_export_preamble,
mysql_view_dependencies_from_rows, mysql_view_dependencies_sql, normalize_export_table_ddl,
record_export_destination_identity, record_export_error, replace_database_export_select_list,
sort_export_views_by_dependencies, split_postgres_export_table_triggers, write_database_export_rows,
BuildDatabaseSqlExportOptions, BuildExportInsertStatementsOptions, DatabaseExportObjectCounts,
DatabaseExportRequest, DdlNormalizeOptions, ExportedTableSql, PostgresExportExtension, PostgresExportSequence,
PostgresExtensionMembers, DATABASE_EXPORT_INSERT_BATCH_SIZE, DATABASE_EXPORT_ROW_LIMIT,
};
use super::{concurrent_metadata_prefetch_allowed, database_export_metadata_prefetch_concurrency};
use crate::connection::AppState;
use crate::models::connection::DatabaseType;
use crate::storage::Storage;
use crate::types::{ObjectInfo, ObjectSourceKind, TableInfo};
use serde_json::{json, Value};
fn table(name: &str, table_type: &str) -> TableInfo {
TableInfo {
name: name.to_string(),
table_type: table_type.to_string(),
comment: None,
parent_schema: None,
parent_name: None,
}
}
fn routine(name: &str, signature: &str) -> ObjectInfo {
ObjectInfo {
name: name.to_string(),
object_type: "FUNCTION".to_string(),
schema: Some("public".to_string()),
valid: None,
signature: Some(signature.to_string()),
custom_type_kind: None,
has_members: None,
comment: None,
created_at: None,
updated_at: None,
parent_schema: None,
parent_name: None,
trigger: None,
xugu_type_members_expandable: None,
}
}
fn export_request(
include_structure: bool,
include_data: bool,
include_objects: bool,
selected_tables: Vec<String>,
) -> DatabaseExportRequest {
DatabaseExportRequest {
export_id: "export-1".to_string(),
connection_id: "connection-1".to_string(),
database: "database-1".to_string(),
schema: "public".to_string(),
file_path: "export.sql".to_string(),
selected_tables,
excluded_tables: Vec::new(),
include_structure,
include_data,
include_objects,
include_create_database: false,
drop_table_if_exists: false,
omit_auto_increment: false,
fail_on_error: false,
snapshot_session_id: None,
batch_size: 1000,
}
}
#[test]
fn export_progress_total_counts_only_requested_object_categories() {
let counts = DatabaseExportObjectCounts {
tables: 2,
views: 1,
sequences: 2,
extensions: 1,
procedures: 1,
functions: 1,
};
let cases = [
("structure", export_request(true, false, false, Vec::new()), 5),
("data", export_request(false, true, false, Vec::new()), 2),
("objects", export_request(false, false, true, Vec::new()), 3),
("all", export_request(true, true, true, Vec::new()), 8),
("nothing", export_request(false, false, false, Vec::new()), 0),
];
for (name, request, expected) in cases {
assert_eq!(database_export_total_objects(&request, &counts), expected, "{name}");
}
}
#[test]
fn mysql_database_export_preamble_preserves_database_settings() {
assert_eq!(
mysql_database_export_preamble("app`db", Some("utf8mb4"), Some("utf8mb4_unicode_ci")),
"CREATE DATABASE IF NOT EXISTS `app``db` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;\nUSE `app``db`;\n"
);
assert_eq!(
mysql_database_export_preamble("app", None, None),
"CREATE DATABASE IF NOT EXISTS `app`;\nUSE `app`;\n"
);
}
#[test]
fn export_progress_total_excludes_schema_routines_for_selected_tables() {
// Counts are already filtered to the selected table/view set before
// workload calculation; schema-wide routines remain intentionally out.
let counts = DatabaseExportObjectCounts {
tables: 1,
views: 1,
sequences: 1,
extensions: 1,
procedures: 4,
functions: 5,
};
let request = export_request(true, true, true, vec!["users".to_string(), "active_users".to_string()]);
assert_eq!(database_export_total_objects(&request, &counts), 4);
}
#[test]
fn postgres_extension_ddl_uses_target_default_version_and_source_schema() {
let extension = PostgresExportExtension { name: "pg_trgm".to_string(), schema: "addons".to_string() };
let ddl = generate_postgres_extension_ddl(&extension);
assert_eq!(ddl, "CREATE EXTENSION IF NOT EXISTS \"pg_trgm\" WITH SCHEMA \"addons\";");
assert!(!ddl.contains("VERSION"));
}
#[test]
fn postgres_extension_member_filter_keeps_user_overload_with_same_name() {
let mut members = PostgresExtensionMembers::default();
members.function_keys.insert(("similarity".to_string(), "text, text".to_string()));
assert!(is_postgres_extension_member_routine(&routine("similarity", "text, text"), &members));
assert!(!is_postgres_extension_member_routine(&routine("similarity", "integer, integer"), &members));
assert!(!is_postgres_extension_member_routine(&routine("user_similarity", "text, text"), &members));
}
#[test]
fn postgres_database_export_defers_table_triggers_without_splitting_function_bodies() {
let ddl = "CREATE TABLE \"public\".\"work_log\" (\n \"id\" bigint,\n \"note\" text DEFAULT ';'::text\n);\n\nCREATE INDEX \"idx_work_log\" ON \"public\".\"work_log\" (\"id\");\n\nCREATE TRIGGER trg_work_log BEFORE INSERT OR UPDATE ON public.work_log FOR EACH ROW EXECUTE FUNCTION fn_work_log_update();";
let (table_ddl, triggers) = split_postgres_export_table_triggers(ddl, DatabaseType::Postgres);
assert!(table_ddl.contains("CREATE TABLE"));
assert!(table_ddl.contains("DEFAULT ';'::text"));
assert!(table_ddl.contains("CREATE INDEX"));
assert!(!table_ddl.contains("CREATE TRIGGER"));
assert_eq!(triggers.len(), 1);
assert!(triggers[0].starts_with("CREATE TRIGGER trg_work_log"));
}
#[test]
fn postgres_database_export_defers_constraint_triggers_only_for_postgres() {
let ddl = "CREATE TABLE \"public\".\"items\" (\"id\" bigint);\nCREATE CONSTRAINT TRIGGER items_check AFTER INSERT ON public.items DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION check_items();";
let (postgres_table, postgres_triggers) = split_postgres_export_table_triggers(ddl, DatabaseType::Postgres);
assert!(!postgres_table.contains("CREATE CONSTRAINT TRIGGER"));
assert_eq!(postgres_triggers.len(), 1);
let (mysql_ddl, mysql_triggers) = split_postgres_export_table_triggers(ddl, DatabaseType::Mysql);
assert_eq!(mysql_ddl, ddl);
assert!(mysql_triggers.is_empty());
}
#[test]
fn concurrent_prefetch_only_allowed_for_multi_connection_pools() {
use crate::connection::PoolKind;
// ChClient::new 只构造 HTTP 客户端,不发起连接
let clickhouse = PoolKind::ClickHouse(crate::db::clickhouse_driver::ChClient::new(
"http://127.0.0.1:1",
None,
None,
std::time::Duration::from_secs(1),
));
assert!(concurrent_metadata_prefetch_allowed(Some(&clickhouse)));
// Agent(JDBC sidecar)请求超时覆盖排队时间,必须回退串行
let agent = PoolKind::agent(crate::db::agent_driver::AgentDriverClient::test_stub());
assert!(!concurrent_metadata_prefetch_allowed(Some(&agent)));
assert!(!concurrent_metadata_prefetch_allowed(None));
}
#[test]
fn postgres_metadata_prefetch_reserves_pool_capacity() {
assert_eq!(database_export_metadata_prefetch_concurrency(DatabaseType::Postgres), 4);
assert_eq!(database_export_metadata_prefetch_concurrency(DatabaseType::Mysql), 8);
}
#[test]
fn filters_export_tables_by_selected_names() {
let tables = vec![table("users", "TABLE"), table("orders", "TABLE"), table("active_users", "VIEW")];
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"]);
}
#[test]
fn keeps_all_export_tables_when_selection_is_empty() {
let tables = vec![table("users", "TABLE"), table("orders", "TABLE")];
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 mysql_export_orders_referenced_views_before_dependents() {
let tables = [table("a_view", "VIEW"), table("z_view", "VIEW")];
let views = tables.iter().collect::<Vec<_>>();
let dependencies =
vec![("a_view".to_string(), "z_view".to_string()), ("z_view".to_string(), "base_table".to_string())];
let sorted = sort_export_views_by_dependencies(&views, &dependencies);
assert_eq!(sorted.iter().map(|view| view.name.as_str()).collect::<Vec<_>>(), vec!["z_view", "a_view"]);
}
#[test]
fn mysql_view_dependency_metadata_is_escaped_and_parsed() {
let sql = mysql_view_dependencies_sql("prod'o");
assert!(sql.contains("VIEW_SCHEMA = 'prod''o'"));
assert!(sql.contains("TABLE_SCHEMA = 'prod''o'"));
let rows = vec![
vec![json!(" a_view "), json!(" z_view ")],
vec![Value::Null, json!("ignored")],
vec![json!(""), json!("ignored")],
];
assert_eq!(mysql_view_dependencies_from_rows(&rows), vec![("a_view".to_string(), "z_view".to_string())]);
}
#[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);
assert_eq!(sql, "DROP TABLE IF EXISTS `users`;");
}
#[test]
fn builds_drop_table_if_exists_without_empty_schema() {
let sql = drop_table_if_exists_sql("users", "", &DatabaseType::Postgres);
assert_eq!(sql, "DROP TABLE IF EXISTS \"users\";");
}
#[test]
fn mysql_export_adds_drop_if_exists_for_views_and_escapes_names() {
let sql = build_database_export_object_source_sql(
DatabaseType::Mysql,
&ObjectSourceKind::View,
"active`rows",
"CREATE VIEW `active``rows` AS SELECT 1",
true,
);
assert_eq!(sql, "DROP VIEW IF EXISTS `active``rows`;\nCREATE VIEW `active``rows` AS SELECT 1;");
}
#[test]
fn mysql_export_adds_drop_if_exists_before_delimited_routines() {
let procedure = build_database_export_object_source_sql(
DatabaseType::Mysql,
&ObjectSourceKind::Procedure,
"refresh_cache",
"CREATE PROCEDURE `refresh_cache`() BEGIN SELECT 1; END",
true,
);
let function = build_database_export_object_source_sql(
DatabaseType::Mysql,
&ObjectSourceKind::Function,
"active_count",
"CREATE FUNCTION `active_count`() RETURNS INT RETURN 1",
true,
);
assert_eq!(
procedure,
"DROP PROCEDURE IF EXISTS `refresh_cache`;\nDELIMITER //\nCREATE PROCEDURE `refresh_cache`() BEGIN SELECT 1; END//\nDELIMITER ;"
);
assert_eq!(
function,
"DROP FUNCTION IF EXISTS `active_count`;\nDELIMITER //\nCREATE FUNCTION `active_count`() RETURNS INT RETURN 1//\nDELIMITER ;"
);
}
#[test]
fn object_drop_option_does_not_change_disabled_or_non_mysql_exports() {
let mysql_without_drop = build_database_export_object_source_sql(
DatabaseType::Mysql,
&ObjectSourceKind::View,
"active_rows",
"CREATE VIEW `active_rows` AS SELECT 1",
false,
);
let postgres_with_drop = build_database_export_object_source_sql(
DatabaseType::Postgres,
&ObjectSourceKind::View,
"active_rows",
"CREATE VIEW active_rows AS SELECT 1",
true,
);
assert_eq!(mysql_without_drop, "CREATE VIEW `active_rows` AS SELECT 1;");
assert_eq!(postgres_with_drop, "CREATE VIEW active_rows AS SELECT 1;");
}
#[test]
fn formats_sql_literals_for_export_inserts() {
assert_eq!(format_export_sql_literal(&Value::Null), "NULL");
assert_eq!(format_export_sql_literal(&json!(42)), "42");
assert_eq!(format_export_sql_literal(&json!(true)), "TRUE");
assert_eq!(format_export_sql_literal(&json!("O'Hara")), "'O''Hara'");
}
#[test]
fn mysql_spatial_export_uses_wkb_constructor_and_preserves_srid() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Mysql),
identifier_quote: None,
schema: None,
table_name: Some("places".to_string()),
qualified_table_name: None,
columns: vec!["location".to_string(), "shape".to_string()],
column_types: vec![Some("point".to_string()), Some("geometry".to_string())],
column_extras: Vec::new(),
rows: vec![vec![
json!("DBX_WKB:4326:0101000000AE47E17A14AE5C4052B81E85EBF34240"),
json!("DBX_WKB:0:0101000000000000000000F03F0000000000000040"),
]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(
statements,
vec!["INSERT INTO `places` (`location`, `shape`) VALUES (ST_GeomFromWKB(0x0101000000AE47E17A14AE5C4052B81E85EBF34240, 4326), ST_GeomFromWKB(0x0101000000000000000000F03F0000000000000040));"]
);
}
#[test]
fn mysql_spatial_export_rejects_markers_for_unknown_or_nonspatial_types() {
let marker = json!("DBX_WKB:4326:0101000000AE47E17A14AE5C4052B81E85EBF34240");
assert!(format_mysql_spatial_export_literal(&marker, Some(DatabaseType::Mysql), None).is_none());
assert!(format_mysql_spatial_export_literal(&marker, Some(DatabaseType::Mysql), Some("varchar")).is_none());
assert!(format_mysql_spatial_export_literal(
&json!("DBX_WKB:4326:0101000000"),
Some(DatabaseType::Mysql),
Some("geometry"),
)
.is_none());
}
#[test]
fn mysql_spatial_export_select_normalizes_geometry_columns_to_wkb_markers() {
let sql = database_export_select_sql(
&["id".to_string(), "location".to_string(), "name".to_string()],
&[Some("int".to_string()), Some("point".to_string()), Some("varchar(32)".to_string())],
"places",
"app",
&DatabaseType::Mysql,
);
assert_eq!(
sql,
"SELECT `id`, CASE WHEN `location` IS NULL THEN NULL ELSE CONCAT('DBX_WKB:', ST_SRID(`location`), ':', HEX(ST_AsWKB(`location`))) END AS `location`, `name` FROM `places`"
);
}
#[test]
fn mysql_spatial_export_keeps_keyset_pagination_when_replacing_select_list() {
let columns = vec!["id".to_string(), "location".to_string()];
let column_types = vec![Some("bigint".to_string()), Some("geometry".to_string())];
let sql = crate::transfer::keyset_pagination_sql_with_identifier_quote(
&columns,
"places",
"app",
&DatabaseType::Mysql,
&["id".to_string()],
&[json!(7)],
1000,
None,
);
let sql = replace_database_export_select_list(sql, &columns, &column_types, &DatabaseType::Mysql);
assert!(sql.starts_with(
"SELECT `id`, CASE WHEN `location` IS NULL THEN NULL ELSE CONCAT('DBX_WKB:', ST_SRID(`location`), ':', HEX(ST_AsWKB(`location`))) END AS `location` FROM `places`"
));
assert!(sql.contains("WHERE `id` > 7"), "sql: {sql}");
assert!(sql.contains("ORDER BY `id` ASC LIMIT 1000"), "sql: {sql}");
}
#[test]
fn database_specific_boolean_export_literals() {
let sqlserver_statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::SqlServer),
identifier_quote: None,
schema: Some("dbo".to_string()),
table_name: Some("flags".to_string()),
qualified_table_name: None,
columns: vec!["typed_true".to_string(), "untyped_false".to_string(), "typed_null".to_string()],
column_types: vec![Some("bit".to_string()), None, Some("bit".to_string())],
column_extras: Vec::new(),
rows: vec![vec![json!(true), json!(false), Value::Null]],
batch_size: Some(10),
})
.unwrap();
let postgres_statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Postgres),
identifier_quote: None,
schema: Some("public".to_string()),
table_name: Some("flags".to_string()),
qualified_table_name: None,
columns: vec!["enabled".to_string(), "disabled".to_string(), "unknown".to_string()],
column_types: Vec::new(),
column_extras: Vec::new(),
rows: vec![vec![json!(true), json!(false), Value::Null]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(
sqlserver_statements,
vec!["INSERT INTO [dbo].[flags] ([typed_true], [untyped_false], [typed_null]) VALUES (1, 0, NULL);"]
);
assert_eq!(
postgres_statements,
vec![
"INSERT INTO \"public\".\"flags\" (\"enabled\", \"disabled\", \"unknown\") VALUES (TRUE, FALSE, NULL);"
]
);
}
#[test]
fn sqlserver_export_prefixes_unicode_string_literals() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::SqlServer),
identifier_quote: None,
schema: Some("dbo".to_string()),
table_name: Some("people".to_string()),
qualified_table_name: None,
columns: vec![
"name".to_string(),
"code".to_string(),
"legacy_note".to_string(),
"alias_name".to_string(),
"plain_text".to_string(),
"missing".to_string(),
],
column_types: vec![
Some("NVARCHAR(255)".to_string()),
Some(" nchar (10) ".to_string()),
Some("ntext".to_string()),
Some("sysname".to_string()),
Some("varchar(255)".to_string()),
Some("nvarchar(20)".to_string()),
],
column_extras: Vec::new(),
rows: vec![vec![
json!("张'三"),
json!("中文"),
json!("旧文本"),
json!("别名"),
json!("plain"),
Value::Null,
]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(
statements,
vec![
"INSERT INTO [dbo].[people] ([name], [code], [legacy_note], [alias_name], [plain_text], [missing]) VALUES (N'张''三', N'中文', N'旧文本', N'别名', 'plain', NULL);"
]
);
}
#[test]
fn mysql_export_inserts_escape_control_characters() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Mysql),
identifier_quote: None,
schema: None,
table_name: Some("notes".to_string()),
qualified_table_name: None,
columns: vec!["body".to_string()],
column_types: vec![Some("text".to_string())],
column_extras: Vec::new(),
rows: vec![vec![json!("line1\nline2\tcol\rend\\slash\0\x1aO'Hara")]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(
statements,
vec!["INSERT INTO `notes` (`body`) VALUES ('line1\\nline2\\tcol\\rend\\\\slash\\0\\ZO''Hara');"]
);
}
#[test]
fn export_insert_batches_split_on_statement_bytes() {
let long_value = "x".repeat(300_000);
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Mysql),
identifier_quote: None,
schema: None,
table_name: Some("payloads".to_string()),
qualified_table_name: None,
columns: vec!["payload".to_string()],
column_types: vec![Some("longtext".to_string())],
column_extras: Vec::new(),
rows: vec![vec![json!(long_value.clone())], vec![json!(long_value)]],
batch_size: Some(100),
})
.unwrap();
assert_eq!(statements.len(), 2);
assert!(statements.iter().all(|statement| statement.matches("INSERT INTO").count() == 1));
}
#[test]
fn sqlserver_export_caps_multi_row_insert_at_1000_rows() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::SqlServer),
identifier_quote: None,
schema: Some("dbo".to_string()),
table_name: Some("items".to_string()),
qualified_table_name: None,
columns: vec!["id".to_string()],
column_types: vec![Some("int".to_string())],
column_extras: Vec::new(),
rows: (0..1001).map(|id| vec![json!(id)]).collect(),
batch_size: Some(2000),
})
.unwrap();
assert_eq!(statements.len(), 2);
assert_eq!(statements[0].matches("), (").count(), 999);
assert_eq!(statements[1].matches("), (").count(), 0);
}
#[test]
fn doris_export_inserts_escape_control_characters() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Doris),
identifier_quote: None,
schema: Some("warehouse".to_string()),
table_name: Some("events".to_string()),
qualified_table_name: None,
columns: vec!["message".to_string()],
column_types: vec![Some("varchar(255)".to_string())],
column_extras: Vec::new(),
rows: vec![vec![json!("first\nsecond\tthird")]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(statements, vec!["INSERT INTO `warehouse`.`events` (`message`) VALUES ('first\\nsecond\\tthird');"]);
}
#[test]
fn postgres_export_inserts_escape_control_characters() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Postgres),
identifier_quote: None,
schema: Some("public".to_string()),
table_name: Some("notes".to_string()),
qualified_table_name: None,
columns: vec!["body".to_string()],
column_types: vec![Some("text".to_string())],
column_extras: Vec::new(),
rows: vec![vec![json!("line1\nline2\tend")]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(statements, vec!["INSERT INTO \"public\".\"notes\" (\"body\") VALUES (E'line1\\nline2\\tend');"]);
}
#[test]
fn postgres_export_inserts_escape_quotes_and_backslashes_without_changing_plain_strings() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Postgres),
identifier_quote: None,
schema: Some("public".to_string()),
table_name: Some("notes".to_string()),
qualified_table_name: None,
columns: vec!["carriage_return".to_string(), "quote".to_string(), "path".to_string(), "plain".to_string()],
column_types: vec![
Some("text".to_string()),
Some("text".to_string()),
Some("text".to_string()),
Some("text".to_string()),
],
column_extras: Vec::new(),
rows: vec![vec![json!("line1\rline2"), json!("O'Hara"), json!(r"C:\tmp"), json!("plain")]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(
statements,
vec![
r#"INSERT INTO "public"."notes" ("carriage_return", "quote", "path", "plain") VALUES (E'line1\rline2', 'O''Hara', E'C:\\tmp', 'plain');"#
]
);
}
#[test]
fn postgres_jsonb_export_preserves_json_escape_sequences() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Postgres),
identifier_quote: None,
schema: Some("public".to_string()),
table_name: Some("events".to_string()),
qualified_table_name: None,
columns: vec!["payload".to_string()],
column_types: vec![Some("jsonb".to_string())],
column_extras: Vec::new(),
rows: vec![vec![json!(r#"{"text":"say \"hi\"","path":"C:\\tmp","quote":"O'Hara"}"#)]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(
statements,
vec![
r#"INSERT INTO "public"."events" ("payload") VALUES (E'{"text":"say \\"hi\\"","path":"C:\\\\tmp","quote":"O''Hara"}');"#
]
);
}
#[test]
fn postgres_vector_export_preserves_pgvector_bracket_literals() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Postgres),
identifier_quote: None,
schema: Some("public".to_string()),
table_name: Some("items".to_string()),
qualified_table_name: None,
columns: vec![
"id".to_string(),
"embedding".to_string(),
"qualified_embedding".to_string(),
"labels".to_string(),
],
column_types: vec![
Some("integer".to_string()),
Some("vector(2)".to_string()),
Some("public.vector".to_string()),
Some("text[]".to_string()),
],
column_extras: Vec::new(),
rows: vec![vec![json!(1), json!([1.2, 3.4]), json!(["5", "6"]), json!(["x", "y"])]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(
statements,
vec![
r#"INSERT INTO "public"."items" ("id", "embedding", "qualified_embedding", "labels") VALUES (1, '[1.2,3.4]', '[5,6]', '{"x","y"}');"#
]
);
}
#[test]
fn builds_batched_insert_statements_for_export() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Mysql),
identifier_quote: None,
schema: None,
table_name: Some("users".to_string()),
qualified_table_name: None,
columns: vec!["id".to_string(), "name".to_string()],
column_types: Vec::new(),
column_extras: Vec::new(),
rows: vec![vec![json!(1), json!("Ada")], vec![json!(2), json!("O'Hara")], vec![json!(3), json!("Linus")]],
batch_size: Some(2),
})
.unwrap();
assert_eq!(
statements,
vec![
"INSERT INTO `users` (`id`, `name`) VALUES (1, 'Ada'), (2, 'O''Hara');",
"INSERT INTO `users` (`id`, `name`) VALUES (3, 'Linus');",
]
);
}
#[test]
fn oracle_export_inserts_use_one_statement_per_row() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Oracle),
identifier_quote: None,
schema: Some("APP".to_string()),
table_name: Some("USERS".to_string()),
qualified_table_name: None,
columns: vec!["ID".to_string(), "NAME".to_string()],
column_types: Vec::new(),
column_extras: Vec::new(),
rows: vec![vec![json!(1), json!("Ada")], vec![json!(2), json!("Linus")]],
batch_size: Some(100),
})
.unwrap();
assert_eq!(
statements,
vec![
"INSERT INTO \"APP\".\"USERS\" (\"ID\", \"NAME\") VALUES (1, 'Ada');",
"INSERT INTO \"APP\".\"USERS\" (\"ID\", \"NAME\") VALUES (2, 'Linus');",
]
);
}
#[test]
fn oracle_export_omits_synthetic_rowid_from_insert_columns() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Oracle),
identifier_quote: None,
schema: Some("APP".to_string()),
table_name: Some("USERS".to_string()),
qualified_table_name: None,
columns: vec!["__DBX_ROWID".to_string(), "ID".to_string(), "NAME".to_string()],
column_types: vec![Some("VARCHAR2".to_string()), Some("NUMBER".to_string()), Some("VARCHAR2".to_string())],
column_extras: Vec::new(),
rows: vec![vec![json!("AAAPr9AAEAAAAGfAAA"), json!(1), json!("Ada")]],
batch_size: Some(100),
})
.unwrap();
assert_eq!(statements, vec!["INSERT INTO \"APP\".\"USERS\" (\"ID\", \"NAME\") VALUES (1, 'Ada');"]);
}
#[test]
fn oceanbase_oracle_export_omits_synthetic_rowid_from_insert_columns() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::OceanbaseOracle),
identifier_quote: None,
schema: Some("APP".to_string()),
table_name: Some("USERS".to_string()),
qualified_table_name: None,
columns: vec!["__DBX_ROWID".to_string(), "ID".to_string(), "NAME".to_string()],
column_types: vec![Some("VARCHAR2".to_string()), Some("NUMBER".to_string()), Some("VARCHAR2".to_string())],
column_extras: Vec::new(),
rows: vec![vec![json!("*AAABk1AAEAAAAAgAAA"), json!(1), json!("Ada")]],
batch_size: Some(100),
})
.unwrap();
assert_eq!(statements, vec!["INSERT INTO \"APP\".\"USERS\" (\"ID\", \"NAME\") VALUES (1, 'Ada');"]);
}
#[test]
fn non_oracle_export_preserves_dbx_rowid_named_column() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Mysql),
identifier_quote: None,
schema: None,
table_name: Some("users".to_string()),
qualified_table_name: None,
columns: vec!["__DBX_ROWID".to_string(), "name".to_string()],
column_types: Vec::new(),
column_extras: Vec::new(),
rows: vec![vec![json!(7), json!("Ada")]],
batch_size: Some(100),
})
.unwrap();
assert_eq!(statements, vec!["INSERT INTO `users` (`__DBX_ROWID`, `name`) VALUES (7, 'Ada');"]);
}
#[test]
fn oracle_date_columns_export_as_date_literals() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Oracle),
identifier_quote: None,
schema: Some("APP".to_string()),
table_name: Some("EVENTS".to_string()),
qualified_table_name: None,
columns: vec!["ID".to_string(), "CREATED_ON".to_string(), "RAW_TEXT".to_string()],
column_types: vec![Some("NUMBER".to_string()), Some("DATE".to_string()), Some("VARCHAR2(64)".to_string())],
column_extras: Vec::new(),
rows: vec![
vec![json!(1), json!("2022-08-25T09:58:43Z"), json!("2022-08-25T09:58:43Z")],
vec![json!(2), json!("2022-08-25T00:00:00Z"), json!("2022-08-25T00:00:00Z")],
],
batch_size: Some(100),
})
.unwrap();
assert_eq!(
statements,
vec![
"INSERT INTO \"APP\".\"EVENTS\" (\"ID\", \"CREATED_ON\", \"RAW_TEXT\") VALUES (1, TO_DATE('2022-08-25 09:58:43', 'YYYY-MM-DD HH24:MI:SS'), '2022-08-25T09:58:43Z');",
"INSERT INTO \"APP\".\"EVENTS\" (\"ID\", \"CREATED_ON\", \"RAW_TEXT\") VALUES (2, DATE '2022-08-25', '2022-08-25T00:00:00Z');",
]
);
}
#[test]
fn oracle_timestamp_columns_export_as_explicit_literals() {
for database_type in [DatabaseType::Oracle, DatabaseType::OceanbaseOracle] {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(database_type),
identifier_quote: None,
schema: Some("APP".to_string()),
table_name: Some("EVENTS".to_string()),
qualified_table_name: None,
columns: vec![
"CREATED_AT".to_string(),
"UPDATED_AT".to_string(),
"RECORDED_AT".to_string(),
"LOCAL_RECORDED_AT".to_string(),
"DRIVER_CREATED_AT".to_string(),
"DRIVER_RECORDED_AT".to_string(),
"RAW_TEXT".to_string(),
"UNTYPED".to_string(),
"INVALID_AT".to_string(),
"SHAPED_INVALID_AT".to_string(),
"NULL_AT".to_string(),
],
column_types: vec![
Some("TIMESTAMP(6)".to_string()),
Some("TIMESTAMP(6)".to_string()),
Some("TIMESTAMP(9) WITH TIME ZONE".to_string()),
Some("TIMESTAMP(6) WITH LOCAL TIME ZONE".to_string()),
Some("TimeStampDTY".to_string()),
Some("TimeStampTZ_DTY".to_string()),
Some("VARCHAR2(64)".to_string()),
None,
Some("TIMESTAMP(6)".to_string()),
Some("TIMESTAMP(6)".to_string()),
Some("TIMESTAMP".to_string()),
],
column_extras: Vec::new(),
rows: vec![vec![
json!("2022-08-25 09:58:43.123456"),
json!("2022-08-26T10:59:44Z"),
json!("2022-08-27T11:00:45.123456789+08:00"),
json!("2022-08-28T12:01:46Z"),
json!("2022-08-29T13:02:47.123456"),
json!("2022-08-30T14:03:48-05:30"),
json!("2022-08-31T15:04:49Z"),
json!("2022-09-01T16:05:50Z"),
json!("not-a-timestamp"),
json!("2022-0'-25T09:58:43Z"),
Value::Null,
]],
batch_size: Some(100),
})
.unwrap();
assert_eq!(
statements,
vec![concat!(
"INSERT INTO \"APP\".\"EVENTS\" (\"CREATED_AT\", \"UPDATED_AT\", \"RECORDED_AT\", ",
"\"LOCAL_RECORDED_AT\", \"DRIVER_CREATED_AT\", \"DRIVER_RECORDED_AT\", \"RAW_TEXT\", ",
"\"UNTYPED\", \"INVALID_AT\", \"SHAPED_INVALID_AT\", \"NULL_AT\") VALUES ",
"(TO_TIMESTAMP('2022-08-25 09:58:43.123456', 'YYYY-MM-DD HH24:MI:SS.FF'), ",
"TO_TIMESTAMP('2022-08-26 10:59:44', 'YYYY-MM-DD HH24:MI:SS'), ",
"TO_TIMESTAMP_TZ('2022-08-27 11:00:45.123456789 +08:00', ",
"'YYYY-MM-DD HH24:MI:SS.FF TZH:TZM'), ",
"TO_TIMESTAMP_TZ('2022-08-28 12:01:46 +00:00', 'YYYY-MM-DD HH24:MI:SS TZH:TZM'), ",
"TO_TIMESTAMP('2022-08-29 13:02:47.123456', 'YYYY-MM-DD HH24:MI:SS.FF'), ",
"TO_TIMESTAMP_TZ('2022-08-30 14:03:48 -05:30', 'YYYY-MM-DD HH24:MI:SS TZH:TZM'), ",
"'2022-08-31T15:04:49Z', '2022-09-01T16:05:50Z', 'not-a-timestamp', ",
"'2022-0''-25T09:58:43Z', NULL);"
)]
);
}
}
#[test]
fn non_oracle_timestamp_exports_keep_existing_literals() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Postgres),
identifier_quote: None,
schema: Some("public".to_string()),
table_name: Some("events".to_string()),
qualified_table_name: None,
columns: vec!["created_at".to_string(), "recorded_at".to_string()],
column_types: vec![Some("timestamp".to_string()), Some("timestamp with time zone".to_string())],
column_extras: Vec::new(),
rows: vec![vec![json!("2022-08-25T09:58:43.123456Z"), json!("2022-08-26T10:59:44+08:00")]],
batch_size: Some(100),
})
.unwrap();
assert_eq!(
statements,
vec![concat!(
"INSERT INTO \"public\".\"events\" (\"created_at\", \"recorded_at\") VALUES ",
"('2022-08-25 09:58:43.123456', '2022-08-26 10:59:44+08:00');"
)]
);
}
#[test]
fn mysql_bit_columns_export_without_quoted_string_values() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Mysql),
identifier_quote: None,
schema: None,
table_name: Some("flags".to_string()),
qualified_table_name: None,
columns: vec!["enabled".to_string(), "mask".to_string(), "label".to_string()],
column_types: vec![Some("bit(1)".to_string()), Some("BIT(4)".to_string()), Some("varchar(20)".to_string())],
column_extras: Vec::new(),
rows: vec![vec![json!("1"), json!("1010"), json!("1010")], vec![json!(false), json!(3), json!("off")]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(
statements,
vec!["INSERT INTO `flags` (`enabled`, `mask`, `label`) VALUES (b'1', b'1010', '1010'), (b'0', 3, 'off');"]
);
}
#[test]
fn dameng_bit_columns_export_as_numeric_literals() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Dameng),
identifier_quote: None,
schema: Some("DBX_TEST".to_string()),
table_name: Some("FLAGS".to_string()),
qualified_table_name: None,
columns: vec!["ENABLED".to_string(), "DELETED".to_string(), "OPTIONAL".to_string()],
column_types: vec![Some("BIT".to_string()), Some("bit".to_string()), Some("BIT".to_string())],
column_extras: Vec::new(),
rows: vec![vec![json!(true), json!(false), Value::Null]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(
statements,
vec!["INSERT INTO \"DBX_TEST\".\"FLAGS\" (\"ENABLED\", \"DELETED\", \"OPTIONAL\") VALUES (1, 0, NULL);"]
);
}
#[test]
fn dameng_strings_export_nul_as_chr_expression() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Dameng),
identifier_quote: None,
schema: Some("DBX_TEST".to_string()),
table_name: Some("NUL_VALUES".to_string()),
qualified_table_name: None,
columns: vec![
"PLAIN".to_string(),
"TRAILING".to_string(),
"LEADING".to_string(),
"MIDDLE".to_string(),
"CONSECUTIVE".to_string(),
"ONLY_NUL".to_string(),
],
column_types: vec![Some("VARCHAR".to_string()); 6],
column_extras: Vec::new(),
rows: vec![vec![
json!("plain"),
json!("eHall\0"),
json!("\0leading"),
json!("left\0right"),
json!("left\0\0right"),
json!("\0"),
]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(
statements,
vec![concat!(
"INSERT INTO \"DBX_TEST\".\"NUL_VALUES\" ",
"(\"PLAIN\", \"TRAILING\", \"LEADING\", \"MIDDLE\", \"CONSECUTIVE\", \"ONLY_NUL\") ",
"VALUES ('plain', 'eHall' || CHR(0), CHR(0) || 'leading', 'left' || CHR(0) || 'right', ",
"'left' || CHR(0) || CHR(0) || 'right', CHR(0));"
)]
);
assert!(!statements[0].contains('\0'));
}
#[test]
fn mysql_export_uses_typed_literals_for_numeric_and_blob_columns() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Mysql),
identifier_quote: None,
schema: None,
table_name: Some("t_test_01".to_string()),
qualified_table_name: None,
columns: vec!["id".to_string(), "f_blob".to_string(), "note".to_string()],
column_types: vec![Some("int".to_string()), Some("blob".to_string()), Some("varchar(64)".to_string())],
column_extras: Vec::new(),
rows: vec![
vec![json!("1"), json!("0x68656c6c6f"), json!("0x68656c6c6f")],
vec![json!("2"), json!("0X"), json!("1")],
],
batch_size: Some(10),
})
.unwrap();
assert_eq!(
statements,
vec![
"INSERT INTO `t_test_01` (`id`, `f_blob`, `note`) VALUES (1, 0x68656c6c6f, '0x68656c6c6f'), (2, X'', '1');"
]
);
}
#[test]
fn temporal_columns_export_without_rfc3339_separator_or_utc_suffix() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Mysql),
identifier_quote: None,
schema: None,
table_name: Some("events".to_string()),
qualified_table_name: None,
columns: vec!["id".to_string(), "created_at".to_string(), "created_on".to_string(), "raw_text".to_string()],
column_types: vec![
Some("int".to_string()),
Some("timestamp".to_string()),
Some("date".to_string()),
Some("varchar(64)".to_string()),
],
column_extras: Vec::new(),
rows: vec![vec![
json!(1),
json!("2026-06-12T10:11:12.123456789Z"),
json!("2026-06-12T10:11:12Z"),
json!("2026-06-12T10:11:12Z"),
]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(
statements,
vec![
"INSERT INTO `events` (`id`, `created_at`, `created_on`, `raw_text`) VALUES (1, '2026-06-12 10:11:12.123456', '2026-06-12', '2026-06-12T10:11:12Z');"
]
);
}
#[test]
fn postgres_timestamptz_export_keeps_timezone_without_rfc3339_t_separator() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Postgres),
identifier_quote: None,
schema: Some("public".to_string()),
table_name: Some("events".to_string()),
qualified_table_name: None,
columns: vec!["recorded_at".to_string(), "local_at".to_string()],
column_types: vec![
Some("timestamp with time zone".to_string()),
Some("timestamp without time zone".to_string()),
],
column_extras: Vec::new(),
rows: vec![vec![json!("2026-06-12T10:11:12Z"), json!("2026-06-12T18:11:12+08:00")]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(
statements,
vec![
"INSERT INTO \"public\".\"events\" (\"recorded_at\", \"local_at\") VALUES ('2026-06-12 10:11:12+00:00', '2026-06-12 18:11:12');"
]
);
}
#[test]
fn sqlserver_rowversion_timestamp_type_is_not_treated_as_datetime() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::SqlServer),
identifier_quote: None,
schema: Some("dbo".to_string()),
table_name: Some("events".to_string()),
qualified_table_name: None,
columns: vec!["row_version".to_string(), "created_at".to_string()],
column_types: vec![Some("timestamp".to_string()), Some("datetime2(3)".to_string())],
column_extras: Vec::new(),
rows: vec![vec![json!("2026-06-12T10:11:12Z"), json!("2026-06-12T10:11:12.1234567Z")]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(
statements,
vec![
"INSERT INTO [dbo].[events] ([row_version], [created_at]) VALUES ('2026-06-12T10:11:12Z', '2026-06-12 10:11:12.123');"
]
);
}
#[test]
fn postgres_tsvector_columns_are_omitted_from_sql_insert_export() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Postgres),
identifier_quote: None,
schema: Some("public".to_string()),
table_name: Some("articles".to_string()),
qualified_table_name: None,
columns: vec!["id".to_string(), "title".to_string(), "search_vector".to_string()],
column_types: vec![Some("integer".to_string()), Some("text".to_string()), Some("tsvector".to_string())],
column_extras: Vec::new(),
rows: vec![vec![json!(1), json!("Hello"), json!("'hello':1A")]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(statements, vec!["INSERT INTO \"public\".\"articles\" (\"id\", \"title\") VALUES (1, 'Hello');"]);
}
#[test]
fn mysql_generated_columns_are_omitted_from_sql_inserts_but_kept_in_ddl() {
let ddl = "CREATE TABLE `orders` (`id` bigint AUTO_INCREMENT, `quantity` int, `unit_price` decimal(10,2), `virtual_total` decimal(10,2) GENERATED ALWAYS AS ((`quantity` * `unit_price`)) VIRTUAL, `stored_total` decimal(10,2) GENERATED ALWAYS AS ((`quantity` * `unit_price`)) STORED);";
let sql = build_database_sql_export(BuildDatabaseSqlExportOptions {
database_name: "shop".to_string(),
exported_at: Some("2026-07-30T00:00:00.000Z".to_string()),
tables: vec![ExportedTableSql {
display_name: "orders".to_string(),
database_type: Some(DatabaseType::Mysql),
identifier_quote: None,
schema: None,
table_name: Some("orders".to_string()),
qualified_table_name: None,
ddl: Some(ddl.to_string()),
columns: vec![
"id".to_string(),
"quantity".to_string(),
"unit_price".to_string(),
"virtual_total".to_string(),
"stored_total".to_string(),
"created_at".to_string(),
],
column_types: vec![
Some("bigint".to_string()),
Some("int".to_string()),
Some("decimal(10,2)".to_string()),
Some("decimal(10,2)".to_string()),
Some("decimal(10,2)".to_string()),
Some("timestamp".to_string()),
],
column_extras: vec![
Some("auto_increment".to_string()),
None,
None,
Some("VIRTUAL GENERATED".to_string()),
Some("stored generated".to_string()),
Some("DEFAULT_GENERATED".to_string()),
],
rows: vec![vec![json!(7), json!(2), json!(3.5), json!(7.0), json!(7.0), json!("2026-07-30 08:00:00")]],
truncated: false,
}],
row_limit_per_table: Some(DATABASE_EXPORT_ROW_LIMIT),
insert_batch_size: Some(DATABASE_EXPORT_INSERT_BATCH_SIZE),
connection_id: None,
database: None,
schema: None,
omit_auto_increment: false,
})
.unwrap();
assert!(sql.contains(ddl));
assert!(sql.contains(
"INSERT INTO `orders` (`id`, `quantity`, `unit_price`, `created_at`) VALUES (7, 2, 3.5, '2026-07-30 08:00:00');"
));
assert!(!sql.contains("INSERT INTO `orders` (`id`, `quantity`, `unit_price`, `virtual_total`"));
}
#[test]
fn mysql_database_export_file_rows_omit_generated_columns() {
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("orders.sql");
let mut file = std::fs::File::create(&path).unwrap();
write_database_export_rows(
&mut file,
&[vec![json!(7), json!(2), json!(7.0), json!("2026-07-30 08:00:00")]],
&["id".to_string(), "quantity".to_string(), "virtual_total".to_string(), "created_at".to_string()],
&[
Some("bigint".to_string()),
Some("int".to_string()),
Some("decimal(10,2)".to_string()),
Some("timestamp".to_string()),
],
&[
Some("auto_increment".to_string()),
None,
Some("VIRTUAL GENERATED".to_string()),
Some("DEFAULT_GENERATED".to_string()),
],
"orders",
"shop",
&DatabaseType::Mysql,
)
.unwrap();
drop(file);
assert_eq!(
std::fs::read_to_string(path).unwrap(),
"INSERT INTO `orders` (`id`, `quantity`, `created_at`) VALUES (7, 2, '2026-07-30 08:00:00');\n\n"
);
}
#[test]
fn dameng_identity_export_inserts_enable_identity_insert() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Dameng),
identifier_quote: None,
schema: Some("SYSDBA".to_string()),
table_name: Some("USERS".to_string()),
qualified_table_name: None,
columns: vec!["ID".to_string(), "NAME".to_string()],
column_types: vec![Some("INT".to_string()), Some("VARCHAR(20)".to_string())],
column_extras: vec![Some("identity".to_string()), None],
rows: vec![vec![json!(1), json!("Ada")]],
batch_size: Some(10),
})
.unwrap();
assert_eq!(
statements,
vec![
"SET IDENTITY_INSERT \"SYSDBA\".\"USERS\" ON;\nINSERT INTO \"SYSDBA\".\"USERS\" (\"ID\", \"NAME\") VALUES (1, 'Ada');\nSET IDENTITY_INSERT \"SYSDBA\".\"USERS\" OFF;"
]
);
}
#[test]
fn builds_database_sql_export_with_ddl_before_data() {
let sql = build_database_sql_export(BuildDatabaseSqlExportOptions {
database_name: "app".to_string(),
exported_at: Some("2026-05-02T00:00:00.000Z".to_string()),
tables: vec![ExportedTableSql {
display_name: "users".to_string(),
database_type: Some(DatabaseType::Mysql),
identifier_quote: None,
schema: None,
table_name: Some("users".to_string()),
qualified_table_name: None,
ddl: Some("CREATE TABLE `users` (`id` int);".to_string()),
columns: vec!["id".to_string()],
column_types: Vec::new(),
column_extras: Vec::new(),
rows: vec![vec![json!(1)]],
truncated: true,
}],
row_limit_per_table: Some(DATABASE_EXPORT_ROW_LIMIT),
insert_batch_size: Some(DATABASE_EXPORT_INSERT_BATCH_SIZE),
connection_id: None,
database: None,
schema: None,
omit_auto_increment: false,
})
.unwrap();
assert_eq!(
sql,
[
"-- DBX database export".to_string(),
"-- Database: app".to_string(),
"-- Exported at: 2026-05-02T00:00:00.000Z".to_string(),
format!("-- Row limit per table: {DATABASE_EXPORT_ROW_LIMIT}"),
String::new(),
"-- Structure for users".to_string(),
"CREATE TABLE `users` (`id` int);".to_string(),
String::new(),
"-- Data for users".to_string(),
format!("-- Exported rows: 1 (truncated at {DATABASE_EXPORT_ROW_LIMIT})"),
"INSERT INTO `users` (`id`) VALUES (1);".to_string(),
String::new(),
]
.join("\n")
);
}
#[test]
fn table_ddl_export_has_one_statement_terminator() {
let ddl = "CREATE TABLE `users` (`id` int);;\n";
assert_eq!(
format_export_table_ddl(ddl, Some(DatabaseType::Mysql), DdlNormalizeOptions::default()),
"CREATE TABLE `users` (`id` int);"
);
assert_eq!(
format_export_table_ddl(
"CREATE TABLE users (id int)",
Some(DatabaseType::Postgres),
DdlNormalizeOptions::default(),
),
"CREATE TABLE users (id int);"
);
}
#[test]
fn omitted_auto_increment_preserves_mysql_line_comment_boundaries() {
let options = DdlNormalizeOptions { omit_auto_increment: true };
assert_eq!(
format_export_table_ddl(
"CREATE TABLE `users` (`id` int) ENGINE=InnoDB -- keep this comment\nAUTO_INCREMENT=5 DEFAULT CHARSET=utf8mb4",
Some(DatabaseType::Mysql),
options,
),
"CREATE TABLE `users` (`id` int) ENGINE=InnoDB -- keep this comment\n DEFAULT CHARSET=utf8mb4;"
);
assert_eq!(
format_export_table_ddl(
"CREATE TABLE `users` (`id` int) ENGINE=InnoDB # keep this comment\r\nAUTO_INCREMENT=5 DEFAULT CHARSET=utf8mb4",
Some(DatabaseType::Mysql),
options,
),
"CREATE TABLE `users` (`id` int) ENGINE=InnoDB # keep this comment\r\n DEFAULT CHARSET=utf8mb4;"
);
}
#[test]
fn omitted_auto_increment_consumes_only_horizontal_separator_whitespace() {
let options = DdlNormalizeOptions { omit_auto_increment: true };
for separator in [" ", "\t"] {
let ddl = format!(
"CREATE TABLE `users` (`id` int) ENGINE=InnoDB{separator}AUTO_INCREMENT=5 DEFAULT CHARSET=utf8mb4"
);
assert_eq!(
format_export_table_ddl(&ddl, Some(DatabaseType::Mysql), options),
"CREATE TABLE `users` (`id` int) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;"
);
}
}
#[test]
fn normalizes_legacy_mysql_row_format_for_export_compatibility() {
let ddl = "CREATE TABLE `wide_table` (\n `payload` varchar(4096) DEFAULT NULL\n) ENGINE=InnoDB DEFAULT CHARSET=utf8 ROW_FORMAT=COMPACT";
let normalized = normalize_export_table_ddl(ddl, Some(DatabaseType::Mysql), DdlNormalizeOptions::default());
assert_eq!(
normalized,
"CREATE TABLE `wide_table` (\n `payload` varchar(4096) DEFAULT NULL\n) ENGINE=InnoDB DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC"
);
}
#[test]
fn normalizes_lowercase_redundant_mysql_row_format_for_export_compatibility() {
let ddl = "CREATE TABLE `wide_table` (`payload` varchar(4096)) engine=InnoDB row_format = redundant";
let normalized = normalize_export_table_ddl(ddl, Some(DatabaseType::Mysql), DdlNormalizeOptions::default());
assert_eq!(normalized, "CREATE TABLE `wide_table` (`payload` varchar(4096)) engine=InnoDB ROW_FORMAT=DYNAMIC");
}
#[test]
fn preserves_non_legacy_or_non_mysql_row_formats() {
let mysql_ddl = "CREATE TABLE `ok` (`payload` text) ENGINE=InnoDB ROW_FORMAT=COMPRESSED";
let postgres_ddl = "CREATE TABLE users (payload text) ROW_FORMAT=COMPACT";
assert_eq!(
normalize_export_table_ddl(mysql_ddl, Some(DatabaseType::Mysql), DdlNormalizeOptions::default()),
mysql_ddl
);
assert_eq!(
normalize_export_table_ddl(postgres_ddl, Some(DatabaseType::Postgres), DdlNormalizeOptions::default()),
postgres_ddl
);
}
fn postgres_sequence(name: &str) -> PostgresExportSequence {
PostgresExportSequence {
name: name.to_string(),
data_type: "integer".to_string(),
start_value: "1".to_string(),
min_value: "1".to_string(),
max_value: "2147483647".to_string(),
increment: "1".to_string(),
cycle: false,
cache_value: "1".to_string(),
last_value: Some("42".to_string()),
owner_table: Some("permissions".to_string()),
owner_column: Some("id".to_string()),
}
}
#[test]
fn postgres_sequence_create_ddl_is_importable_before_table_ddl() {
let ddl = generate_postgres_sequence_create_ddl(&postgres_sequence("permissions_id_seq"), "public");
assert_eq!(
ddl,
[
"CREATE SEQUENCE IF NOT EXISTS \"public\".\"permissions_id_seq\"",
" AS integer",
" START WITH 1",
" INCREMENT BY 1",
" MINVALUE 1",
" MAXVALUE 2147483647",
" CACHE 1",
" NO CYCLE",
]
.join("\n")
);
}
#[test]
fn postgres_sequence_owner_and_setval_sql_are_qualified() {
let sequence = postgres_sequence("permissions_id_seq");
assert_eq!(
generate_postgres_sequence_owner_ddl(&sequence, "public").as_deref(),
Some("ALTER SEQUENCE \"public\".\"permissions_id_seq\" OWNED BY \"public\".\"permissions\".\"id\"")
);
assert_eq!(
generate_postgres_sequence_setval_sql(&sequence, "public").as_deref(),
Some(
"SELECT setval('\"public\".\"permissions_id_seq\"', GREATEST(COALESCE(MAX(\"id\"), 42), 42), true) FROM \"public\".\"permissions\""
)
);
}
#[test]
fn strict_exports_return_object_errors_instead_of_writing_error_comments() {
let path = std::env::temp_dir().join(format!("dbx-strict-export-{}.sql", uuid::Uuid::new_v4()));
let mut file = std::fs::File::create(&path).unwrap();
let result = record_export_error(&mut file, true, "exporting table users: permission denied".to_string());
drop(file);
assert_eq!(result.unwrap_err(), "exporting table users: permission denied");
assert!(std::fs::read_to_string(&path).unwrap().is_empty());
let _ = std::fs::remove_file(path);
}
async fn test_app_state(scratch_dir: &std::path::Path) -> AppState {
let storage = Storage::open(&scratch_dir.join("storage.db")).await.unwrap();
AppState::new(storage)
}
// Regression tests for #6327: create_dir_all on every export run is unsafe
// when the destination is on a removable/network drive, because a mount
// that is temporarily gone at write time would be silently recreated on
// the local root filesystem, and the backup would "succeed" while writing
// to the wrong disk.
#[tokio::test]
async fn ensure_export_destination_dir_creates_a_never_before_seen_local_directory() {
let scratch = std::env::temp_dir().join(format!("dbx-export-dest-new-{}", uuid::Uuid::new_v4()));
std::fs::create_dir_all(&scratch).unwrap();
let state = test_app_state(&scratch).await;
let destination = scratch.join("backups").join("mydb");
assert!(!destination.exists());
ensure_export_destination_dir(&state, &destination)
.await
.expect("first-time local directory should be created");
assert!(destination.is_dir());
let _ = std::fs::remove_dir_all(&scratch);
}
#[tokio::test]
async fn ensure_export_destination_dir_refuses_to_recreate_a_destination_that_disappeared() {
let scratch = std::env::temp_dir().join(format!("dbx-export-dest-vanished-{}", uuid::Uuid::new_v4()));
std::fs::create_dir_all(&scratch).unwrap();
let state = test_app_state(&scratch).await;
// Simulates a previously configured, already-used backup destination
// (e.g. on an external or network drive) by successfully exporting to
// it once first.
let destination = scratch.join("mounted-drive").join("mydb");
ensure_export_destination_dir(&state, &destination).await.expect("initial export should create the directory");
assert!(destination.is_dir());
// Simulates the mount disappearing (unplugged drive, unmounted share,
// etc.) before the next run.
std::fs::remove_dir_all(scratch.join("mounted-drive")).unwrap();
assert!(!destination.exists());
let result = ensure_export_destination_dir(&state, &destination).await;
assert!(result.is_err(), "a destination that existed before should not be silently recreated");
assert!(!destination.exists(), "the backup directory must not be resurrected on the wrong filesystem");
let _ = std::fs::remove_dir_all(&scratch);
}
// Regression test for review feedback on #6327: `ensure_export_destination_dir`
// only remembered a destination *after* a successful export, so a mount
// that was present when a schedule was configured but vanished before its
// very first run looked identical to a brand-new local folder (no
// recorded state either way) and got silently recreated on the local
// disk. `record_export_destination_identity` closes that gap by letting
// the schedule-configuration flow record the destination eagerly.
#[tokio::test]
async fn record_export_destination_identity_protects_a_mount_that_vanishes_before_its_first_run() {
let scratch = std::env::temp_dir().join(format!("dbx-export-dest-eager-{}", uuid::Uuid::new_v4()));
std::fs::create_dir_all(&scratch).unwrap();
let state = test_app_state(&scratch).await;
// Simulates the user picking an already-mounted external/network
// drive in the schedule editor and saving the schedule -- the
// directory exists at configuration time, but no export has run yet.
let destination = scratch.join("mounted-drive").join("mydb");
std::fs::create_dir_all(&destination).unwrap();
record_export_destination_identity(&state, &destination)
.await
.expect("configuring the schedule should succeed");
// The mount disappears before the scheduler ever runs this schedule
// for the first time.
std::fs::remove_dir_all(scratch.join("mounted-drive")).unwrap();
assert!(!destination.exists());
let result = ensure_export_destination_dir(&state, &destination).await;
assert!(result.is_err(), "a mount that was recorded at configuration time must not be silently recreated");
assert!(!destination.exists(), "the backup directory must not be resurrected on the wrong filesystem");
let _ = std::fs::remove_dir_all(&scratch);
}
#[tokio::test]
async fn record_export_destination_identity_rejects_a_directory_that_disappeared_before_save() {
let scratch = std::env::temp_dir().join(format!("dbx-export-dest-eager-new-{}", uuid::Uuid::new_v4()));
std::fs::create_dir_all(&scratch).unwrap();
let state = test_app_state(&scratch).await;
// The schedule editor only accepts an existing directory. If that
// directory disappears before the save request reaches the backend,
// the schedule must not be persisted without a recorded identity.
let destination = scratch.join("backups").join("mydb");
assert!(!destination.exists());
let error = record_export_destination_identity(&state, &destination)
.await
.expect_err("a missing scheduled destination must be rejected");
assert!(error.contains("does not exist or is not a directory"));
assert!(!destination.exists());
let _ = std::fs::remove_dir_all(&scratch);
}
// Regression test for review feedback on #6327: the directory identity
// check and the later `File::create` are separate operations, so the
// mount can disappear and be replaced by something else at the same path
// in between. `export_destination_identity_mismatch` is the comparison
// `export_database_sql_core` runs against the handle it actually opened;
// exercised directly here since reproducing a real cross-filesystem swap
// mid-write is not something a portable unit test can simulate.
#[test]
fn export_destination_identity_mismatch_detects_a_changed_device() {
assert!(export_destination_identity_mismatch(Some(1), Some(2)));
assert!(!export_destination_identity_mismatch(Some(1), Some(1)));
assert!(
!export_destination_identity_mismatch(None, Some(2)),
"an unknown expected device has nothing to compare against"
);
assert!(
export_destination_identity_mismatch(Some(1), None),
"an opened file with unknown identity must not bypass a known expected device"
);
assert!(!export_destination_identity_mismatch(None, None));
}
// Regression test for review feedback on #6327: the non-Unix path used
// to report no device identity at all, so replacing a Windows drive or
// mount at the same path went undetected. This only runs on native
// Windows (this repo's CI has no Windows job that executes `cargo test`,
// only `cargo check`, so it is exercised locally by Windows contributors
// and by the compile-check itself).
#[cfg(windows)]
#[test]
fn export_destination_device_id_for_path_and_open_file_agree_on_windows() {
let scratch = std::env::temp_dir().join(format!("dbx-export-dest-win-{}", uuid::Uuid::new_v4()));
std::fs::create_dir_all(&scratch).unwrap();
let dir_dev = super::export_destination_device_id_for_path(&scratch);
assert!(dir_dev.is_some(), "a real local directory should report a volume serial number");
let file_path = scratch.join("probe.txt");
let file = std::fs::File::create(&file_path).unwrap();
let file_dev = super::export_destination_device_id_for_file(&file);
drop(file);
assert_eq!(dir_dev, file_dev, "a file and its parent directory must resolve to the same volume");
let _ = std::fs::remove_dir_all(&scratch);
}
#[test]
fn kingbase_mysql_compat_export_insert_uses_backtick_identifiers() {
let statements = build_export_insert_statements(BuildExportInsertStatementsOptions {
database_type: Some(DatabaseType::Kingbase),
identifier_quote: Some("`".to_string()),
schema: Some("audit-schema".to_string()),
table_name: Some("events".to_string()),
qualified_table_name: None,
columns: vec!["id".to_string(), "event_type".to_string()],
column_types: vec![None, None],
column_extras: vec![None, None],
rows: vec![vec![json!(1), json!("login")]],
batch_size: Some(100),
})
.unwrap();
assert_eq!(statements, vec!["INSERT INTO `audit-schema`.`events` (`id`, `event_type`) VALUES (1, 'login');"]);
}
}