2302 lines
85 KiB
Rust
2302 lines
85 KiB
Rust
use std::collections::HashMap;
|
|
|
|
use minijinja::{context, Environment};
|
|
use serde::{Deserialize, Serialize};
|
|
|
|
use crate::correction::JointCorrectionPlan;
|
|
use crate::data_compare::DataCompareFromTablesPreparation;
|
|
use crate::dml_binding::BindingEngine;
|
|
use crate::models::connection::DatabaseType;
|
|
use crate::schema_diff::{
|
|
generate_schema_sync_sql, ColumnCompatibilityWarning, RollbackGraph, SchemaDiffPreparation, TableDiff,
|
|
};
|
|
use crate::sql_dialect::descriptor::{
|
|
DialectCapabilityDescriptor, DialectKind, CAP_CREATE_OR_REPLACE, CAP_IF_NOT_EXISTS,
|
|
};
|
|
use crate::state_persistence::StateBackend;
|
|
use crate::types::TableInfo;
|
|
|
|
// ============================================================================
|
|
// 9.2: Jinja2 Dialect-Aware Template Engine
|
|
// ============================================================================
|
|
|
|
const SCHEMA_SYNC_TEMPLATE: &str = r#"-- ============================================================
|
|
-- Database Schema Synchronization Script
|
|
-- Generated: {{ generated_at }}
|
|
-- Source Dialect: {{ source_dialect }}
|
|
-- Target Dialect: {{ target_dialect }}
|
|
-- Database Type: {{ db_type }}
|
|
{% if target_schema %}-- Target Schema: {{ target_schema }}{% endif %}
|
|
-- Strategy: {{ idempotent_strategy }}
|
|
-- ============================================================
|
|
|
|
{% if pre_checks %}{% for check in pre_checks %}
|
|
-- Pre-check: {{ check }}
|
|
{% endfor %}
|
|
{% endif %}
|
|
|
|
{{ header_comment | safe }}
|
|
|
|
{% if sync_sql %}
|
|
{{ sync_sql | safe }}
|
|
{% endif %}
|
|
|
|
{% if rollback_sql %}
|
|
-- ============================================================
|
|
-- Rollback Script
|
|
-- ============================================================
|
|
{{ rollback_sql | safe }}
|
|
{% endif %}
|
|
|
|
-- End of synchronization script
|
|
"#;
|
|
|
|
const JOINT_ORCHESTRATION_TEMPLATE: &str = r#"-- ============================================================
|
|
-- Joint Structure-Data Synchronization Script
|
|
-- Generated: {{ generated_at }}
|
|
-- Strategy: {{ strategy }}
|
|
-- Schema Steps: {{ schema_diff_count }}
|
|
-- Data Steps: {{ data_diff_count }}
|
|
-- Total Steps: {{ total_steps }}
|
|
-- ============================================================
|
|
|
|
{% if pre_checks %}{% for check in pre_checks %}
|
|
-- Pre-check: {{ check }}
|
|
{% endfor %}
|
|
{% endif %}
|
|
|
|
{% for step in steps %}
|
|
-- ------------------------------------------------------------------
|
|
-- Step {{ loop.index }}: {{ step.step_type }} [{{ step.risk_level }}]
|
|
-- {{ step.description }}
|
|
{% if step.table_name %}-- Table: {{ step.table_name }}{% endif %}
|
|
{% if step.depends_on %}-- Depends on: {{ step.depends_on }}{% endif %}
|
|
-- ------------------------------------------------------------------
|
|
{{ step.sql }}
|
|
|
|
{% if step.rollback_sql %}
|
|
-- Rollback:
|
|
{{ step.rollback_sql }}
|
|
{% endif %}
|
|
|
|
{% if include_checkpoints and step.step_type == "Checkpoint" %}
|
|
-- <<< CHECKPOINT: {{ step.description }} >>>
|
|
{% endif %}
|
|
{% endfor %}
|
|
|
|
-- End of joint synchronization script
|
|
"#;
|
|
|
|
const BATCH_TEMPLATE: &str = r#"-- ============================================================
|
|
-- Batch {{ batch_id }} / {{ total_batches }}
|
|
-- Description: {{ description }}
|
|
-- Estimated Rows: {{ estimated_rows }}
|
|
-- ============================================================
|
|
|
|
{% if pre_checks %}{% for check in pre_checks %}
|
|
-- Pre-check: {{ check }}
|
|
{% endfor %}
|
|
{% endif %}
|
|
|
|
{{ sql | safe }}
|
|
|
|
{% if rollback_sql %}
|
|
-- Rollback for batch {{ batch_id }}:
|
|
{{ rollback_sql | safe }}
|
|
{% endif %}
|
|
|
|
-- End of batch {{ batch_id }}
|
|
-- <<< CHECKPOINT: batch_{{ batch_id }}_complete >>>
|
|
"#;
|
|
|
|
pub struct ScriptTemplateEngine {
|
|
env: Environment<'static>,
|
|
}
|
|
|
|
impl Default for ScriptTemplateEngine {
|
|
fn default() -> Self {
|
|
Self::new()
|
|
}
|
|
}
|
|
|
|
impl ScriptTemplateEngine {
|
|
pub fn new() -> Self {
|
|
let mut env = Environment::new();
|
|
env.add_template("schema_sync", SCHEMA_SYNC_TEMPLATE).expect("schema_sync template");
|
|
env.add_template("joint_orchestration", JOINT_ORCHESTRATION_TEMPLATE).expect("joint_orchestration template");
|
|
env.add_template("batch", BATCH_TEMPLATE).expect("batch template");
|
|
Self { env }
|
|
}
|
|
|
|
pub fn from_dialect_yaml(dialect: &crate::sql_dialect::dialect_yaml::ScriptTemplatesYaml) -> Self {
|
|
let mut engine = Self::new();
|
|
if let Some(tmpl) = &dialect.schema_sync {
|
|
let leaked: &'static str = Box::leak(tmpl.clone().into_boxed_str());
|
|
engine.env.add_template("schema_sync", leaked).expect("schema_sync from yaml");
|
|
}
|
|
if let Some(tmpl) = &dialect.joint_orchestration {
|
|
let leaked: &'static str = Box::leak(tmpl.clone().into_boxed_str());
|
|
engine.env.add_template("joint_orchestration", leaked).expect("joint_orchestration from yaml");
|
|
}
|
|
if let Some(tmpl) = &dialect.batch {
|
|
let leaked: &'static str = Box::leak(tmpl.clone().into_boxed_str());
|
|
engine.env.add_template("batch", leaked).expect("batch from yaml");
|
|
}
|
|
engine
|
|
}
|
|
|
|
pub fn render_schema_sync(
|
|
&self,
|
|
sync_sql: &str,
|
|
rollback_sql: Option<&str>,
|
|
db_type: &str,
|
|
source_dialect: &str,
|
|
target_dialect: &str,
|
|
idempotent_strategy: &str,
|
|
target_schema: Option<&str>,
|
|
header_comment: &str,
|
|
pre_checks: &[String],
|
|
) -> Result<String, String> {
|
|
let tmpl = self.env.get_template("schema_sync").map_err(|e| e.to_string())?;
|
|
tmpl.render(context! {
|
|
generated_at => chrono::Utc::now().to_rfc3339(),
|
|
db_type,
|
|
source_dialect,
|
|
target_dialect,
|
|
idempotent_strategy,
|
|
target_schema,
|
|
header_comment,
|
|
sync_sql,
|
|
rollback_sql => rollback_sql.unwrap_or(""),
|
|
pre_checks,
|
|
})
|
|
.map_err(|e| e.to_string())
|
|
}
|
|
|
|
pub fn render_joint_script(
|
|
&self,
|
|
plan: &JointCorrectionPlan,
|
|
include_checkpoints: bool,
|
|
pre_checks: &[String],
|
|
) -> Result<String, String> {
|
|
let tmpl = self.env.get_template("joint_orchestration").map_err(|e| e.to_string())?;
|
|
let steps: Vec<HashMap<String, String>> = plan
|
|
.steps
|
|
.iter()
|
|
.map(|s| {
|
|
let mut m = HashMap::new();
|
|
m.insert("step_type".to_string(), format!("{:?}", s.step_type));
|
|
m.insert("risk_level".to_string(), format!("{:?}", s.risk_level));
|
|
m.insert("description".to_string(), s.description.clone());
|
|
m.insert("sql".to_string(), s.sql.clone());
|
|
m.insert("rollback_sql".to_string(), s.rollback_sql.clone().unwrap_or_default());
|
|
m.insert("table_name".to_string(), s.table_name.clone().unwrap_or_default());
|
|
m.insert(
|
|
"depends_on".to_string(),
|
|
if s.depends_on.is_empty() {
|
|
String::new()
|
|
} else {
|
|
s.depends_on.iter().map(|d| d.to_string()).collect::<Vec<_>>().join(", ")
|
|
},
|
|
);
|
|
m
|
|
})
|
|
.collect();
|
|
|
|
tmpl.render(context! {
|
|
generated_at => chrono::Utc::now().to_rfc3339(),
|
|
strategy => format!("{:?}", plan.strategy),
|
|
schema_diff_count => plan.schema_diff_count,
|
|
data_diff_count => plan.data_diff_count,
|
|
total_steps => plan.steps.len(),
|
|
steps,
|
|
include_checkpoints,
|
|
pre_checks,
|
|
})
|
|
.map_err(|e| e.to_string())
|
|
}
|
|
|
|
pub fn render_batch(
|
|
&self,
|
|
batch_id: usize,
|
|
total_batches: usize,
|
|
description: &str,
|
|
sql: &str,
|
|
rollback_sql: Option<&str>,
|
|
estimated_rows: Option<u64>,
|
|
pre_checks: &[String],
|
|
) -> Result<String, String> {
|
|
let tmpl = self.env.get_template("batch").map_err(|e| e.to_string())?;
|
|
tmpl.render(context! {
|
|
batch_id,
|
|
total_batches,
|
|
description,
|
|
estimated_rows => estimated_rows.unwrap_or(0),
|
|
sql,
|
|
rollback_sql => rollback_sql.unwrap_or(""),
|
|
pre_checks,
|
|
})
|
|
.map_err(|e| e.to_string())
|
|
}
|
|
}
|
|
|
|
// ============================================================================
|
|
// 9.3: Adaptive Idempotent Strategy
|
|
// ============================================================================
|
|
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
|
#[serde(rename_all = "snake_case")]
|
|
pub enum IdempotentStrategy {
|
|
IfNotExists,
|
|
CreateOrReplace,
|
|
ConditionalCheck,
|
|
None,
|
|
}
|
|
|
|
impl IdempotentStrategy {
|
|
pub fn label(&self) -> &'static str {
|
|
match self {
|
|
Self::IfNotExists => "if_not_exists",
|
|
Self::CreateOrReplace => "create_or_replace",
|
|
Self::ConditionalCheck => "conditional_check",
|
|
Self::None => "none",
|
|
}
|
|
}
|
|
}
|
|
|
|
fn dialect_supports(descriptor: Option<&DialectCapabilityDescriptor>, flag: u64) -> bool {
|
|
descriptor.is_some_and(|d| d.flags & flag != 0)
|
|
}
|
|
|
|
pub fn select_strategy(
|
|
_dialect_kind: Option<DialectKind>,
|
|
descriptor: Option<&DialectCapabilityDescriptor>,
|
|
) -> IdempotentStrategy {
|
|
if dialect_supports(descriptor, CAP_IF_NOT_EXISTS) {
|
|
IdempotentStrategy::IfNotExists
|
|
} else if dialect_supports(descriptor, CAP_CREATE_OR_REPLACE) {
|
|
IdempotentStrategy::CreateOrReplace
|
|
} else {
|
|
IdempotentStrategy::ConditionalCheck
|
|
}
|
|
}
|
|
|
|
fn strip_leading_comments(s: &str) -> &str {
|
|
let mut pos = 0;
|
|
for line in s.lines() {
|
|
let t = line.trim();
|
|
if t.is_empty() || t.starts_with("--") {
|
|
pos += line.len() + 1; // +1 for newline
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
if pos >= s.len() {
|
|
""
|
|
} else {
|
|
s[pos..].trim_start()
|
|
}
|
|
}
|
|
|
|
pub fn apply_idempotent_strategy(sql: &str, db_type: DatabaseType, strategy: IdempotentStrategy) -> String {
|
|
if strategy == IdempotentStrategy::None || sql.trim().is_empty() {
|
|
return sql.to_string();
|
|
}
|
|
|
|
let statements = split_sql_statements(sql);
|
|
let mut result = Vec::new();
|
|
|
|
for stmt in statements {
|
|
let trimmed = stmt.trim();
|
|
if trimmed.is_empty() {
|
|
result.push(stmt);
|
|
continue;
|
|
}
|
|
|
|
let stripped = strip_leading_comments(trimmed);
|
|
if stripped.is_empty() {
|
|
result.push(stmt);
|
|
continue;
|
|
}
|
|
|
|
let leading_comment =
|
|
if stripped.len() < trimmed.len() { &trimmed[..trimmed.len() - stripped.len()] } else { "" };
|
|
|
|
let idempotent_body = match strategy {
|
|
IdempotentStrategy::IfNotExists => wrap_if_not_exists(stripped, db_type),
|
|
IdempotentStrategy::CreateOrReplace => wrap_create_or_replace(stripped, db_type),
|
|
IdempotentStrategy::ConditionalCheck => wrap_conditional_check(stripped, db_type),
|
|
IdempotentStrategy::None => trimmed.to_string(),
|
|
};
|
|
|
|
if leading_comment.is_empty() || idempotent_body == trimmed {
|
|
result.push(idempotent_body);
|
|
} else {
|
|
result.push(format!("{leading_comment}{idempotent_body}"));
|
|
}
|
|
}
|
|
|
|
result.join("\n")
|
|
}
|
|
|
|
fn split_sql_statements(sql: &str) -> Vec<String> {
|
|
let mut statements = Vec::new();
|
|
let mut current = String::new();
|
|
let mut in_string = false;
|
|
let mut string_char = ' ';
|
|
let mut chars = sql.chars().peekable();
|
|
|
|
while let Some(ch) = chars.next() {
|
|
if in_string {
|
|
current.push(ch);
|
|
if ch == string_char {
|
|
// SQL escaped quote: '' or "" stays inside the string.
|
|
if chars.peek() == Some(&string_char) {
|
|
current.push(chars.next().unwrap());
|
|
} else {
|
|
in_string = false;
|
|
}
|
|
}
|
|
} else if ch == '\'' || ch == '"' {
|
|
in_string = true;
|
|
string_char = ch;
|
|
current.push(ch);
|
|
} else if ch == ';' {
|
|
current.push(ch);
|
|
statements.push(std::mem::take(&mut current));
|
|
} else {
|
|
current.push(ch);
|
|
}
|
|
}
|
|
|
|
if !current.trim().is_empty() {
|
|
statements.push(current);
|
|
}
|
|
|
|
statements
|
|
}
|
|
|
|
fn upper_first_word(s: &str) -> &str {
|
|
let trimmed = s.trim_start();
|
|
trimmed.split_whitespace().next().unwrap_or("")
|
|
}
|
|
|
|
fn wrap_if_not_exists(sql: &str, db_type: DatabaseType) -> String {
|
|
use crate::sql_dialect::ddl_profile::profile_for;
|
|
let profile = profile_for(db_type);
|
|
let upper = sql.trim_start().to_uppercase();
|
|
|
|
if upper.starts_with("CREATE TABLE") {
|
|
if !profile.create_table_if_not_exists {
|
|
return sql.to_string();
|
|
}
|
|
if upper.contains("IF NOT EXISTS") {
|
|
return sql.to_string();
|
|
}
|
|
let idx = sql.find("CREATE TABLE").unwrap_or(0) + "CREATE TABLE".len();
|
|
let (prefix, suffix) = sql.split_at(idx);
|
|
format!("{prefix} IF NOT EXISTS{suffix}")
|
|
} else if upper.starts_with("CREATE UNIQUE INDEX") || upper.starts_with("CREATE INDEX") {
|
|
if !profile.create_index_if_not_exists {
|
|
return sql.to_string();
|
|
}
|
|
if upper.contains("IF NOT EXISTS") {
|
|
return sql.to_string();
|
|
}
|
|
// PostgreSQL/MySQL: CREATE [UNIQUE] INDEX IF NOT EXISTS name ON ...
|
|
// (not CREATE IF NOT EXISTS INDEX ...)
|
|
let keyword = if upper.starts_with("CREATE UNIQUE INDEX") { "CREATE UNIQUE INDEX" } else { "CREATE INDEX" };
|
|
let idx = upper.find(keyword).unwrap_or(0) + keyword.len();
|
|
let (prefix, suffix) = sql.split_at(idx);
|
|
format!("{prefix} IF NOT EXISTS{suffix}")
|
|
} else if upper.starts_with("DROP TABLE") {
|
|
if upper.contains("IF EXISTS") {
|
|
return sql.to_string();
|
|
}
|
|
let idx = sql.find("DROP TABLE").unwrap_or(0) + "DROP TABLE".len();
|
|
let (prefix, suffix) = sql.split_at(idx);
|
|
format!("{prefix} IF EXISTS{suffix}")
|
|
} else if upper.starts_with("DROP INDEX") {
|
|
if upper.contains("IF EXISTS") {
|
|
return sql.to_string();
|
|
}
|
|
let idx = sql.find("DROP INDEX").unwrap_or(0) + "DROP INDEX".len();
|
|
let (prefix, suffix) = sql.split_at(idx);
|
|
format!("{prefix} IF EXISTS{suffix}")
|
|
} else if upper.starts_with("DROP SEQUENCE") || upper.starts_with("DROP FUNCTION") {
|
|
if upper.contains("IF EXISTS") {
|
|
return sql.to_string();
|
|
}
|
|
let pos = if upper.starts_with("DROP SEQUENCE") {
|
|
sql.find("DROP SEQUENCE").unwrap_or(0) + "DROP SEQUENCE".len()
|
|
} else {
|
|
sql.find("DROP FUNCTION").unwrap_or(0) + "DROP FUNCTION".len()
|
|
};
|
|
let (prefix, suffix) = sql.split_at(pos);
|
|
format!("{prefix} IF EXISTS{suffix}")
|
|
} else {
|
|
sql.to_string()
|
|
}
|
|
}
|
|
|
|
fn wrap_create_or_replace(sql: &str, db_type: DatabaseType) -> String {
|
|
use crate::sql_dialect::ddl_profile::profile_for;
|
|
let profile = profile_for(db_type);
|
|
let upper = sql.trim_start().to_uppercase();
|
|
|
|
if upper.starts_with("CREATE VIEW") {
|
|
if upper.contains("OR REPLACE") {
|
|
return sql.to_string();
|
|
}
|
|
let idx = sql.find("CREATE").unwrap_or(0) + "CREATE".len();
|
|
let (prefix, suffix) = sql.split_at(idx);
|
|
format!("{prefix} OR REPLACE{suffix}")
|
|
} else if upper.starts_with("CREATE FUNCTION") || upper.starts_with("CREATE OR REPLACE FUNCTION") {
|
|
if upper.contains("OR REPLACE") || upper.contains("IF NOT EXISTS") {
|
|
return sql.to_string();
|
|
}
|
|
if profile.create_function_or_replace {
|
|
let idx = sql.find("CREATE").unwrap_or(0) + "CREATE".len();
|
|
let (prefix, suffix) = sql.split_at(idx);
|
|
format!("{prefix} OR REPLACE{suffix}")
|
|
} else {
|
|
sql.to_string()
|
|
}
|
|
} else {
|
|
sql.to_string()
|
|
}
|
|
}
|
|
|
|
fn wrap_conditional_check(sql: &str, db_type: DatabaseType) -> String {
|
|
use crate::sql_dialect::ddl_profile::profile_for;
|
|
let profile = profile_for(db_type);
|
|
let trimmed = sql.trim().trim_end_matches(';').trim_end();
|
|
let upper = trimmed.to_uppercase();
|
|
let first_word = upper_first_word(trimmed);
|
|
|
|
if first_word == "CREATE" && upper.contains("TABLE") {
|
|
// Prefer native IF NOT EXISTS — never append bare DDL after a SELECT CASE
|
|
// (SELECT does not execute the string payload).
|
|
if profile.create_table_if_not_exists {
|
|
return wrap_if_not_exists(sql, db_type);
|
|
}
|
|
let table_identifier = extract_object_identifier(trimmed, "TABLE");
|
|
let table_name = extract_table_name(trimmed, "TABLE");
|
|
let ddl_literal = trimmed.replace('\'', "''");
|
|
return match db_type {
|
|
DatabaseType::Postgres
|
|
| DatabaseType::Gaussdb
|
|
| DatabaseType::OpenGauss
|
|
| DatabaseType::Kingbase
|
|
| DatabaseType::Highgo
|
|
| DatabaseType::Vastbase
|
|
| DatabaseType::Uxdb
|
|
| DatabaseType::Redshift => {
|
|
let (schema_name, table_name) = postgres_table_identity(table_identifier);
|
|
let schema_predicate = schema_name.map_or_else(
|
|
|| "table_schema = current_schema()".to_string(),
|
|
|schema| format!("table_schema = '{}'", schema.replace('\'', "''")),
|
|
);
|
|
let table_literal = table_name.replace('\'', "''");
|
|
format!(
|
|
"DO $dbx_idempotent$\nBEGIN\n IF NOT EXISTS (\n SELECT 1 FROM information_schema.tables\n WHERE {schema_predicate}\n AND table_name = '{table_literal}'\n ) THEN\n EXECUTE $dbx_ddl${trimmed}$dbx_ddl$;\n END IF;\nEND\n$dbx_idempotent$;"
|
|
)
|
|
}
|
|
DatabaseType::Oracle
|
|
| DatabaseType::Dameng
|
|
| DatabaseType::OceanbaseOracle
|
|
| DatabaseType::Yashandb
|
|
| DatabaseType::Xugu
|
|
| DatabaseType::Iris => {
|
|
// ORA-00955: name is already used by an existing object
|
|
format!(
|
|
"BEGIN\n EXECUTE IMMEDIATE '{ddl_literal}';\nEXCEPTION\n WHEN OTHERS THEN\n IF SQLCODE != -955 THEN RAISE; END IF;\nEND;"
|
|
)
|
|
}
|
|
_ => {
|
|
format!("-- ConditionalCheck: run only if table {table_name} does not exist\n{trimmed};")
|
|
}
|
|
};
|
|
}
|
|
|
|
if first_word == "DROP" {
|
|
let rebuilt = format!("{trimmed};");
|
|
let guarded = wrap_if_not_exists(&rebuilt, db_type);
|
|
if guarded != rebuilt && guarded.contains("IF EXISTS") {
|
|
return guarded;
|
|
}
|
|
let obj_type = if upper.contains("TABLE") {
|
|
"TABLE"
|
|
} else if upper.contains("INDEX") {
|
|
"INDEX"
|
|
} else {
|
|
""
|
|
};
|
|
if !obj_type.is_empty() {
|
|
let obj_name = extract_table_name(trimmed, obj_type);
|
|
return format!("-- Conditional: only execute if {obj_type} {obj_name} exists\n{trimmed};");
|
|
}
|
|
}
|
|
|
|
sql.to_string()
|
|
}
|
|
|
|
fn extract_object_identifier<'a>(sql: &'a str, keyword: &str) -> &'a str {
|
|
let upper = sql.to_uppercase();
|
|
let search = format!(" {} ", keyword);
|
|
if let Some(pos) = upper.find(&search) {
|
|
let mut after = sql[pos + search.len()..].trim();
|
|
if after.to_ascii_uppercase().starts_with("IF NOT EXISTS ") {
|
|
after = after["IF NOT EXISTS ".len()..].trim_start();
|
|
}
|
|
|
|
let mut quote = None;
|
|
let mut end = after.len();
|
|
for (index, ch) in after.char_indices() {
|
|
match quote {
|
|
Some('"') if ch == '"' => quote = None,
|
|
Some('`') if ch == '`' => quote = None,
|
|
Some(']') if ch == ']' => quote = None,
|
|
Some(_) => {}
|
|
None if ch == '"' || ch == '`' => quote = Some(ch),
|
|
None if ch == '[' => quote = Some(']'),
|
|
None if ch.is_whitespace() || ch == '(' || ch == ';' => {
|
|
end = index;
|
|
break;
|
|
}
|
|
None => {}
|
|
}
|
|
}
|
|
after[..end].trim()
|
|
} else {
|
|
"unknown"
|
|
}
|
|
}
|
|
|
|
fn extract_table_name<'a>(sql: &'a str, keyword: &str) -> &'a str {
|
|
extract_object_identifier(sql, keyword).trim_matches('"').trim_matches('`').trim_matches('[').trim_matches(']')
|
|
}
|
|
|
|
fn postgres_table_identity(identifier: &str) -> (Option<String>, String) {
|
|
let mut quote = false;
|
|
let mut separator = None;
|
|
let chars: Vec<(usize, char)> = identifier.char_indices().collect();
|
|
let mut index = 0;
|
|
while index < chars.len() {
|
|
let (byte_index, ch) = chars[index];
|
|
if ch == '"' {
|
|
if quote && chars.get(index + 1).is_some_and(|(_, next)| *next == '"') {
|
|
index += 1;
|
|
} else {
|
|
quote = !quote;
|
|
}
|
|
} else if ch == '.' && !quote {
|
|
separator = Some(byte_index);
|
|
}
|
|
index += 1;
|
|
}
|
|
|
|
let (schema, table) = separator
|
|
.map_or((None, identifier), |separator| (Some(&identifier[..separator]), &identifier[separator + 1..]));
|
|
(schema.map(postgres_identifier_value), postgres_identifier_value(table))
|
|
}
|
|
|
|
fn postgres_identifier_value(identifier: &str) -> String {
|
|
let identifier = identifier.trim();
|
|
if identifier.starts_with('"') && identifier.ends_with('"') && identifier.len() >= 2 {
|
|
identifier[1..identifier.len() - 1].replace("\"\"", "\"")
|
|
} else {
|
|
identifier.to_lowercase()
|
|
}
|
|
}
|
|
|
|
// ============================================================================
|
|
// 9.4: Rollback Script Generator
|
|
// ============================================================================
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct RollbackScriptOptions {
|
|
pub db_type: DatabaseType,
|
|
pub target_schema: Option<String>,
|
|
pub cascade_delete: bool,
|
|
pub include_header: bool,
|
|
pub include_safety_checks: bool,
|
|
pub idempotent_strategy: IdempotentStrategy,
|
|
}
|
|
|
|
impl Default for RollbackScriptOptions {
|
|
fn default() -> Self {
|
|
Self {
|
|
db_type: DatabaseType::Postgres,
|
|
target_schema: None,
|
|
cascade_delete: false,
|
|
include_header: true,
|
|
include_safety_checks: true,
|
|
idempotent_strategy: IdempotentStrategy::IfNotExists,
|
|
}
|
|
}
|
|
}
|
|
|
|
pub fn generate_rollback_script(rollback_graph: &RollbackGraph, options: &RollbackScriptOptions) -> String {
|
|
let mut lines = Vec::new();
|
|
|
|
if options.include_header {
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push("-- ROLLBACK SCRIPT".to_string());
|
|
lines.push(format!("-- Generated: {}", chrono::Utc::now().to_rfc3339()));
|
|
lines.push(format!("-- Database Type: {:?}", options.db_type));
|
|
if let Some(schema) = &options.target_schema {
|
|
lines.push(format!("-- Target Schema: {schema}"));
|
|
}
|
|
lines.push(format!("-- Forward Nodes: {}", rollback_graph.forward_nodes.len()));
|
|
lines.push(format!("-- Rollback Nodes: {}", rollback_graph.rollback_nodes.len()));
|
|
lines.push(format!("-- Consistent: {}", rollback_graph.is_consistent));
|
|
if !rollback_graph.consistency_issues.is_empty() {
|
|
lines.push("-- WARNING: Consistency issues detected:".to_string());
|
|
for issue in &rollback_graph.consistency_issues {
|
|
lines.push(format!("-- - {issue}"));
|
|
}
|
|
}
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push(String::new());
|
|
}
|
|
|
|
if options.include_safety_checks {
|
|
lines.push("-- SAFETY: Review this rollback script before execution.".to_string());
|
|
lines.push("-- Ensure the state captured in the rollback graph matches current database state.".to_string());
|
|
lines.push(String::new());
|
|
}
|
|
|
|
let rollback_diffs: Vec<TableDiff> = rollback_graph
|
|
.rollback_nodes
|
|
.iter()
|
|
.map(|n| {
|
|
let mut diff = n.table_diff.clone();
|
|
diff.sync_sql = None;
|
|
diff
|
|
})
|
|
.collect();
|
|
|
|
let sync_sql = generate_schema_sync_sql(
|
|
&rollback_diffs,
|
|
&[],
|
|
&[],
|
|
&[],
|
|
&[],
|
|
options.db_type,
|
|
options.target_schema.as_deref(),
|
|
options.cascade_delete,
|
|
None,
|
|
&[],
|
|
);
|
|
|
|
let idempotent_sql = apply_idempotent_strategy(&sync_sql, options.db_type, options.idempotent_strategy);
|
|
|
|
lines.push(idempotent_sql);
|
|
lines.join("\n")
|
|
}
|
|
|
|
pub fn generate_reverse_diff_rollback(
|
|
forward_sync_sql: &str,
|
|
db_type: DatabaseType,
|
|
schema_diff: &SchemaDiffPreparation,
|
|
options: &RollbackScriptOptions,
|
|
) -> String {
|
|
if let Some(graph) = &schema_diff.rollback_graph {
|
|
return generate_rollback_script(graph, options);
|
|
}
|
|
|
|
let mut lines = Vec::new();
|
|
if options.include_header {
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push("-- ROLLBACK SCRIPT (from reverse diff)".to_string());
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push(String::new());
|
|
}
|
|
|
|
let _ = forward_sync_sql;
|
|
|
|
if let Some(rollback_sql) = &schema_diff.rollback_sync_sql {
|
|
lines.push(apply_idempotent_strategy(rollback_sql, db_type, options.idempotent_strategy));
|
|
} else {
|
|
lines.push("-- No rollback SQL available. Rollback must be designed manually.".to_string());
|
|
}
|
|
|
|
lines.join("\n")
|
|
}
|
|
|
|
// ============================================================================
|
|
// 9.5: Structure-Data Joint Orchestration
|
|
// ============================================================================
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct JointScriptOptions {
|
|
pub strategy: JointScriptStrategy,
|
|
pub include_checkpoints: bool,
|
|
pub include_rollback: bool,
|
|
pub shadow_table_switch: bool,
|
|
pub idempotent_strategy: IdempotentStrategy,
|
|
pub db_type: DatabaseType,
|
|
pub target_schema: Option<String>,
|
|
}
|
|
|
|
impl Default for JointScriptOptions {
|
|
fn default() -> Self {
|
|
Self {
|
|
strategy: JointScriptStrategy::StructureFirst,
|
|
include_checkpoints: true,
|
|
include_rollback: true,
|
|
shadow_table_switch: false,
|
|
idempotent_strategy: IdempotentStrategy::IfNotExists,
|
|
db_type: DatabaseType::Postgres,
|
|
target_schema: None,
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
|
#[serde(rename_all = "snake_case")]
|
|
pub enum JointScriptStrategy {
|
|
StructureFirst,
|
|
DataFirst,
|
|
Interleaved,
|
|
ShadowTable,
|
|
}
|
|
|
|
pub fn generate_joint_script(
|
|
schema_diff: &SchemaDiffPreparation,
|
|
data_compare: Option<&DataCompareFromTablesPreparation>,
|
|
options: &JointScriptOptions,
|
|
) -> String {
|
|
let _engine = ScriptTemplateEngine::new();
|
|
let mut lines: Vec<String> = Vec::new();
|
|
|
|
let header = format!(
|
|
"-- Joint Structure-Data Synchronization Script\n-- Strategy: {:?}\n-- Generated: {}\n-- Database: {:?}\n",
|
|
options.strategy,
|
|
chrono::Utc::now().to_rfc3339(),
|
|
options.db_type
|
|
);
|
|
lines.push(header);
|
|
|
|
if let Some(schema) = &options.target_schema {
|
|
lines.push(format!("-- Target Schema: {schema}"));
|
|
}
|
|
lines.push(String::new());
|
|
|
|
match options.strategy {
|
|
JointScriptStrategy::StructureFirst => {
|
|
lines.push("-- PHASE 1: Structure Changes".to_string());
|
|
lines.push(String::new());
|
|
let schema_sql =
|
|
apply_idempotent_strategy(&schema_diff.sync_sql, options.db_type, options.idempotent_strategy);
|
|
lines.push(schema_sql);
|
|
lines.push(String::new());
|
|
|
|
if let Some(dc) = data_compare {
|
|
lines.push("-- PHASE 2: Data Synchronization".to_string());
|
|
lines.push(String::new());
|
|
if !dc.sync_sql.is_empty() {
|
|
lines.push(dc.sync_sql.clone());
|
|
}
|
|
lines.push(String::new());
|
|
}
|
|
|
|
if options.include_checkpoints {
|
|
lines.push("-- <<< CHECKPOINT: structure_data_sync_complete >>>".to_string());
|
|
lines.push(String::new());
|
|
}
|
|
}
|
|
JointScriptStrategy::DataFirst => {
|
|
lines.push("-- PHASE 1: Data Synchronization".to_string());
|
|
lines.push(String::new());
|
|
if let Some(dc) = data_compare {
|
|
if !dc.sync_sql.is_empty() {
|
|
lines.push(dc.sync_sql.clone());
|
|
}
|
|
lines.push(String::new());
|
|
}
|
|
|
|
lines.push("-- PHASE 2: Structure Changes".to_string());
|
|
lines.push(String::new());
|
|
let schema_sql =
|
|
apply_idempotent_strategy(&schema_diff.sync_sql, options.db_type, options.idempotent_strategy);
|
|
lines.push(schema_sql);
|
|
lines.push(String::new());
|
|
}
|
|
JointScriptStrategy::Interleaved => {
|
|
let table_diffs: Vec<&TableDiff> =
|
|
schema_diff.diffs.iter().filter(|d| d.diff_type != "unchanged" && d.diff_type != "none").collect();
|
|
|
|
for (i, td) in table_diffs.iter().enumerate() {
|
|
lines.push(format!("-- Step {}: Processing table {}", i + 1, td.name));
|
|
lines.push(String::new());
|
|
|
|
if let Some(dc) = data_compare {
|
|
if dc.sync_sql.contains(&td.name) {
|
|
lines.push(format!("-- Data sync for {}", td.name));
|
|
lines.push(dc.sync_sql.clone());
|
|
lines.push(String::new());
|
|
}
|
|
}
|
|
|
|
if let Some(sync_sql) = &td.sync_sql {
|
|
let idempotent = apply_idempotent_strategy(sync_sql, options.db_type, options.idempotent_strategy);
|
|
lines.push(format!("-- Schema sync for {}", td.name));
|
|
lines.push(idempotent);
|
|
lines.push(String::new());
|
|
}
|
|
}
|
|
}
|
|
JointScriptStrategy::ShadowTable => {
|
|
lines.push(generate_shadow_table_script(schema_diff, data_compare, options));
|
|
}
|
|
}
|
|
|
|
if options.include_rollback {
|
|
if let Some(rollback_sql) = &schema_diff.rollback_sync_sql {
|
|
lines.push(String::new());
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push("-- ROLLBACK SECTION".to_string());
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push(apply_idempotent_strategy(rollback_sql, options.db_type, options.idempotent_strategy));
|
|
}
|
|
}
|
|
|
|
lines.push(String::new());
|
|
lines.push("-- End of joint synchronization script".to_string());
|
|
|
|
lines.join("\n")
|
|
}
|
|
|
|
fn generate_shadow_table_script(
|
|
schema_diff: &SchemaDiffPreparation,
|
|
data_compare: Option<&DataCompareFromTablesPreparation>,
|
|
options: &JointScriptOptions,
|
|
) -> String {
|
|
let mut lines = Vec::new();
|
|
|
|
lines.push("-- Shadow Table Switch Strategy".to_string());
|
|
lines.push("-- This strategy creates shadow tables, syncs data, then swaps.".to_string());
|
|
lines.push(String::new());
|
|
|
|
for diff in &schema_diff.diffs {
|
|
if diff.diff_type == "added" {
|
|
lines.push(format!("-- New table {}: create directly (no shadow needed)", diff.name));
|
|
if let Some(sync_sql) = &diff.sync_sql {
|
|
lines.push(apply_idempotent_strategy(sync_sql, options.db_type, options.idempotent_strategy));
|
|
lines.push(String::new());
|
|
}
|
|
} else if diff.diff_type == "modified" {
|
|
let shadow_name = format!("_shadow_{}", diff.name);
|
|
lines.push(format!("-- Shadow table strategy for: {}", diff.name));
|
|
lines.push(String::new());
|
|
|
|
lines.push(format!("-- Step 1: Create shadow table {}", shadow_name));
|
|
if let Some(ddl) = &diff.ddl {
|
|
let shadow_ddl = ddl.replace(&diff.name, &shadow_name);
|
|
lines.push(format!("{shadow_ddl};"));
|
|
lines.push(String::new());
|
|
}
|
|
|
|
lines.push("-- Step 2: Copy data to shadow table".to_string());
|
|
lines.push(format!("INSERT INTO {shadow_name} SELECT * FROM {};", diff.name));
|
|
lines.push(String::new());
|
|
|
|
if let Some(dc) = data_compare {
|
|
if dc.sync_sql.contains(&diff.name) {
|
|
lines.push("-- Step 2b: Apply data sync on shadow table".to_string());
|
|
lines.push(dc.sync_sql.clone());
|
|
lines.push(String::new());
|
|
}
|
|
}
|
|
|
|
lines.push("-- Step 3: Begin transaction and swap tables".to_string());
|
|
lines.push("BEGIN;".to_string());
|
|
lines.push(format!("ALTER TABLE {} RENAME TO _old_{};", diff.name, diff.name));
|
|
lines.push(format!("ALTER TABLE {shadow_name} RENAME TO {};", diff.name));
|
|
|
|
if options.include_rollback {
|
|
lines.push("-- Rollback point: rename _old_ back if needed".to_string());
|
|
}
|
|
|
|
lines.push("COMMIT;".to_string());
|
|
lines.push(String::new());
|
|
|
|
lines.push("-- Step 4: Drop old table (after verification)".to_string());
|
|
lines.push(format!("DROP TABLE IF EXISTS _old_{};", diff.name));
|
|
lines.push(String::new());
|
|
}
|
|
}
|
|
|
|
lines.join("\n")
|
|
}
|
|
|
|
// ============================================================================
|
|
// 9.6: Checkpoint Resume & Large Table Batching
|
|
// ============================================================================
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct BatchConfig {
|
|
pub max_statements_per_batch: usize,
|
|
pub max_estimated_rows_per_batch: u64,
|
|
pub enable_checkpoints: bool,
|
|
pub checkpoint_key_prefix: String,
|
|
}
|
|
|
|
impl Default for BatchConfig {
|
|
fn default() -> Self {
|
|
Self {
|
|
max_statements_per_batch: 100,
|
|
max_estimated_rows_per_batch: 1_000_000,
|
|
enable_checkpoints: true,
|
|
checkpoint_key_prefix: "batch_sync".to_string(),
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct BatchInfo {
|
|
pub batch_id: usize,
|
|
pub description: String,
|
|
pub sql: String,
|
|
pub rollback_sql: Option<String>,
|
|
pub estimated_rows: Option<u64>,
|
|
pub table_names: Vec<String>,
|
|
pub checkpoint_key: Option<String>,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct BatchState {
|
|
pub batch_id: usize,
|
|
pub total_batches: usize,
|
|
pub completed_batches: Vec<usize>,
|
|
pub current_batch: Option<usize>,
|
|
pub status: String,
|
|
pub started_at: String,
|
|
pub updated_at: String,
|
|
}
|
|
|
|
impl BatchState {
|
|
pub fn new(total_batches: usize) -> Self {
|
|
let now = chrono::Utc::now().to_rfc3339();
|
|
Self {
|
|
batch_id: 0,
|
|
total_batches,
|
|
completed_batches: Vec::new(),
|
|
current_batch: None,
|
|
status: "created".to_string(),
|
|
started_at: now.clone(),
|
|
updated_at: now,
|
|
}
|
|
}
|
|
|
|
pub fn checkpoint_key(config: &BatchConfig) -> String {
|
|
format!("{}_state", config.checkpoint_key_prefix)
|
|
}
|
|
}
|
|
|
|
pub struct BatchController {
|
|
config: BatchConfig,
|
|
engine: ScriptTemplateEngine,
|
|
}
|
|
|
|
impl BatchController {
|
|
pub fn new(config: BatchConfig) -> Self {
|
|
Self { config, engine: ScriptTemplateEngine::new() }
|
|
}
|
|
|
|
pub fn build_batches(
|
|
&self,
|
|
sync_sql: &str,
|
|
rollback_sql: Option<&str>,
|
|
table_infos: &[TableInfo],
|
|
) -> Vec<BatchInfo> {
|
|
let statements = split_sql_statements(sync_sql);
|
|
if statements.is_empty() {
|
|
return Vec::new();
|
|
}
|
|
|
|
let total = statements.len();
|
|
let batch_size = self.config.max_statements_per_batch.max(1);
|
|
let num_batches = total.div_ceil(batch_size);
|
|
let engine = &self.engine;
|
|
|
|
let mut batches = Vec::new();
|
|
|
|
for i in 0..num_batches {
|
|
let start = i * batch_size;
|
|
let end = ((i + 1) * batch_size).min(total);
|
|
let batch_sql = statements[start..end].join("\n");
|
|
|
|
let table_names: Vec<String> = statements[start..end]
|
|
.iter()
|
|
.filter_map(|s| {
|
|
let upper = s.trim_start().to_uppercase();
|
|
let tbl = if upper.contains("ALTER TABLE") {
|
|
extract_from_alter(s)
|
|
} else if upper.contains("CREATE TABLE") || upper.contains("DROP TABLE") {
|
|
Some(extract_table_name(s, "TABLE").to_string())
|
|
} else {
|
|
None
|
|
};
|
|
tbl
|
|
})
|
|
.collect();
|
|
|
|
let checkpoint_key = if self.config.enable_checkpoints {
|
|
Some(format!("{}_batch_{}", self.config.checkpoint_key_prefix, i + 1))
|
|
} else {
|
|
None
|
|
};
|
|
|
|
let description = if num_batches > 1 {
|
|
format!("Batch {}/{}: {} tables ({}-{})", i + 1, num_batches, table_names.len(), start + 1, end)
|
|
} else {
|
|
format!("Full sync: {} statements", total)
|
|
};
|
|
|
|
let rendered = engine
|
|
.render_batch(i + 1, num_batches, &description, &batch_sql, None, None, &[])
|
|
.unwrap_or_else(|_| batch_sql.clone());
|
|
|
|
batches.push(BatchInfo {
|
|
batch_id: i + 1,
|
|
description,
|
|
sql: rendered,
|
|
rollback_sql: rollback_sql.map(|s| s.to_string()),
|
|
estimated_rows: None,
|
|
table_names,
|
|
checkpoint_key,
|
|
});
|
|
}
|
|
|
|
let _ = table_infos;
|
|
batches
|
|
}
|
|
|
|
pub async fn save_checkpoint(&self, backend: &dyn StateBackend, state: &BatchState) -> Result<(), String> {
|
|
let key = BatchState::checkpoint_key(&self.config);
|
|
let data = serde_json::to_vec(state).map_err(|e| e.to_string())?;
|
|
backend.save(&key, &data).await
|
|
}
|
|
|
|
pub async fn load_checkpoint(&self, backend: &dyn StateBackend) -> Result<Option<BatchState>, String> {
|
|
let key = BatchState::checkpoint_key(&self.config);
|
|
let data = backend.load(&key).await?;
|
|
match data {
|
|
Some(bytes) => {
|
|
let state: BatchState = serde_json::from_slice(&bytes).map_err(|e| e.to_string())?;
|
|
Ok(Some(state))
|
|
}
|
|
None => Ok(None),
|
|
}
|
|
}
|
|
|
|
pub async fn delete_checkpoint(&self, backend: &dyn StateBackend) -> Result<(), String> {
|
|
let key = BatchState::checkpoint_key(&self.config);
|
|
backend.delete(&key).await
|
|
}
|
|
|
|
pub fn get_next_batch<'a>(&self, state: &BatchState, batches: &'a [BatchInfo]) -> Option<&'a BatchInfo> {
|
|
let next_id = state.completed_batches.last().map(|id| id + 1).unwrap_or(1);
|
|
batches.iter().find(|b| b.batch_id == next_id)
|
|
}
|
|
|
|
pub fn mark_batch_complete(&self, state: &mut BatchState, batch_id: usize) {
|
|
if !state.completed_batches.contains(&batch_id) {
|
|
state.completed_batches.push(batch_id);
|
|
}
|
|
state.current_batch = None;
|
|
state.updated_at = chrono::Utc::now().to_rfc3339();
|
|
if state.completed_batches.len() >= state.total_batches {
|
|
state.status = "completed".to_string();
|
|
}
|
|
}
|
|
}
|
|
|
|
fn extract_from_alter(sql: &str) -> Option<String> {
|
|
let upper = sql.to_uppercase();
|
|
if let Some(pos) = upper.find("ALTER TABLE ") {
|
|
let after = &sql[pos + "ALTER TABLE ".len()..].trim();
|
|
let end = after.find(|c: char| c.is_whitespace() || c == '(' || c == ';').unwrap_or(after.len());
|
|
let name = after[..end].trim().trim_matches('"').trim_matches('`').trim_matches('[').trim_matches(']');
|
|
Some(name.to_string())
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
|
|
// ============================================================================
|
|
// Convenience Functions
|
|
// ============================================================================
|
|
|
|
pub fn generate_enhanced_sync_sql(
|
|
schema_diff: &SchemaDiffPreparation,
|
|
db_type: DatabaseType,
|
|
_target_schema: Option<&str>,
|
|
strategy: IdempotentStrategy,
|
|
) -> String {
|
|
let raw_sql = &schema_diff.sync_sql;
|
|
if raw_sql.is_empty() {
|
|
return String::new();
|
|
}
|
|
apply_idempotent_strategy(raw_sql, db_type, strategy)
|
|
}
|
|
|
|
pub fn generate_enhanced_rollback_sql(
|
|
schema_diff: &SchemaDiffPreparation,
|
|
db_type: DatabaseType,
|
|
target_schema: Option<&str>,
|
|
cascade_delete: bool,
|
|
strategy: IdempotentStrategy,
|
|
) -> Option<String> {
|
|
if let Some(graph) = &schema_diff.rollback_graph {
|
|
let options = RollbackScriptOptions {
|
|
db_type,
|
|
target_schema: target_schema.map(|s| s.to_string()),
|
|
cascade_delete,
|
|
include_header: true,
|
|
include_safety_checks: true,
|
|
idempotent_strategy: strategy,
|
|
};
|
|
Some(generate_rollback_script(graph, &options))
|
|
} else {
|
|
schema_diff.rollback_sync_sql.as_ref().map(|sql| apply_idempotent_strategy(sql, db_type, strategy))
|
|
}
|
|
}
|
|
|
|
pub fn generate_complete_script(
|
|
schema_diff: &SchemaDiffPreparation,
|
|
data_compare: Option<&DataCompareFromTablesPreparation>,
|
|
db_type: DatabaseType,
|
|
target_schema: Option<&str>,
|
|
strategy: IdempotentStrategy,
|
|
) -> String {
|
|
let mut lines = Vec::new();
|
|
|
|
let dialect_kind = DialectKind::from_database_type(db_type);
|
|
let dialect_name = dialect_kind.label();
|
|
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push("-- Complete Database Synchronization Script".to_string());
|
|
lines.push(format!("-- Generated: {}", chrono::Utc::now().to_rfc3339()));
|
|
lines.push(format!("-- Dialect: {dialect_name}"));
|
|
lines.push(format!("-- Idempotent Strategy: {:?}", strategy));
|
|
if let Some(schema) = target_schema {
|
|
lines.push(format!("-- Target Schema: {schema}"));
|
|
}
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push(String::new());
|
|
|
|
let sync_sql = apply_idempotent_strategy(&schema_diff.sync_sql, db_type, strategy);
|
|
|
|
if !sync_sql.is_empty() {
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push("-- SECTION 1: Structure Changes".to_string());
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push(String::new());
|
|
lines.push(sync_sql);
|
|
lines.push(String::new());
|
|
}
|
|
|
|
let dml_clean = generate_all_dml_clean_sql(schema_diff, 0.8);
|
|
if !dml_clean.is_empty() {
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push("-- SECTION 1b: DML Pre-Clean (lossy type mappings)".to_string());
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push(String::new());
|
|
for stmt in &dml_clean {
|
|
lines.push(format!("{stmt};"));
|
|
}
|
|
lines.push(String::new());
|
|
}
|
|
|
|
if let Some(perm_sql) = &schema_diff.permission_sync_sql {
|
|
if !perm_sql.is_empty() {
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push("-- SECTION 2: Permission Changes".to_string());
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push(String::new());
|
|
lines.push(apply_idempotent_strategy(perm_sql, db_type, strategy));
|
|
lines.push(String::new());
|
|
}
|
|
}
|
|
|
|
if let Some(dc) = data_compare {
|
|
if !dc.sync_sql.is_empty() {
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push("-- SECTION 3: Data Synchronization".to_string());
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push(String::new());
|
|
if let Some(level) = &dc.degradation_level {
|
|
lines.push(format!("-- Data Quality: degradation_level={level}"));
|
|
}
|
|
if let Some(confidence) = dc.confidence_score {
|
|
lines.push(format!("-- Confidence Score: {confidence:.2}"));
|
|
}
|
|
lines.push(String::new());
|
|
lines.push(dc.sync_sql.clone());
|
|
lines.push(String::new());
|
|
}
|
|
}
|
|
|
|
if let Some(rollback_sql) = &schema_diff.rollback_sync_sql {
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push("-- SECTION 4: Rollback Script".to_string());
|
|
lines.push("-- ============================================================".to_string());
|
|
lines.push(String::new());
|
|
lines.push(apply_idempotent_strategy(rollback_sql, db_type, strategy));
|
|
lines.push(String::new());
|
|
}
|
|
|
|
lines.push("-- End of complete synchronization script".to_string());
|
|
lines.join("\n")
|
|
}
|
|
|
|
// ============================================================================
|
|
// Phase 14.6: DML Clean — pre-transform SQL for lossy type mappings
|
|
// ============================================================================
|
|
|
|
/// Generate DML clean SQL from compatibility warnings.
|
|
/// Produces structure-data pre-clean statements (e.g. UPDATE ... SET ... = NULL).
|
|
pub fn generate_dml_clean_sql(
|
|
warnings: &[ColumnCompatibilityWarning],
|
|
table_name: &str,
|
|
fidelity_threshold: f64,
|
|
) -> Vec<String> {
|
|
let mut statements = Vec::new();
|
|
let mut seen = std::collections::HashSet::new();
|
|
|
|
for warning in warnings {
|
|
let result = BindingEngine::bind(
|
|
&warning.source_type,
|
|
&warning.target_type,
|
|
table_name,
|
|
&warning.column_name,
|
|
fidelity_threshold,
|
|
);
|
|
|
|
if let Some(sql) = &result.pre_transform_sql {
|
|
if seen.insert(sql.clone()) {
|
|
statements.push(sql.clone());
|
|
}
|
|
}
|
|
}
|
|
|
|
statements
|
|
}
|
|
|
|
/// Generate DML clean SQL for ALL tables in a SchemaDiffPreparation.
|
|
/// Iterates table diffs → modified columns → checks type compatibility → binds DML.
|
|
pub fn generate_all_dml_clean_sql(schema_diff: &SchemaDiffPreparation, fidelity_threshold: f64) -> Vec<String> {
|
|
let mut all_statements = Vec::new();
|
|
|
|
for diff in &schema_diff.diffs {
|
|
let table_name = &diff.name;
|
|
if let Some(columns) = &diff.columns {
|
|
for col_diff in columns {
|
|
if col_diff.diff_type == "modified" {
|
|
if let (Some(src), Some(tgt)) = (&col_diff.source, &col_diff.target) {
|
|
let result = BindingEngine::bind(
|
|
&src.data_type,
|
|
&tgt.data_type,
|
|
table_name,
|
|
&col_diff.name,
|
|
fidelity_threshold,
|
|
);
|
|
if let Some(sql) = result.pre_transform_sql {
|
|
all_statements.push(sql);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
all_statements.dedup();
|
|
all_statements
|
|
}
|
|
|
|
// ============================================================================
|
|
// Phase 15.7: Lock Timeout Injection
|
|
// ============================================================================
|
|
|
|
/// Return dialect-appropriate SET lock_timeout statement.
|
|
pub fn lock_timeout_statement(db_type: DatabaseType) -> Option<&'static str> {
|
|
crate::sql_dialect::ddl_profile::profile_for(db_type).lock_timeout_sql
|
|
}
|
|
|
|
// ============================================================================
|
|
// Tests
|
|
// ============================================================================
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::correction::*;
|
|
use crate::schema_diff::{ColumnDiff, DiffNode, RollbackGraph, SchemaDiffPreparation, TableDiff};
|
|
use crate::sql_dialect::descriptor::{
|
|
DialectCapabilityDescriptor, DialectKind, CAP_CREATE_OR_REPLACE, CAP_CREATE_TABLE, CAP_IF_NOT_EXISTS,
|
|
};
|
|
use crate::types::{ColumnInfo, TableInfo};
|
|
|
|
fn make_table_info(name: &str) -> TableInfo {
|
|
TableInfo {
|
|
name: name.to_string(),
|
|
table_type: "table".to_string(),
|
|
comment: None,
|
|
parent_schema: None,
|
|
parent_name: None,
|
|
}
|
|
}
|
|
|
|
fn make_column_info(name: &str, data_type: &str) -> ColumnInfo {
|
|
ColumnInfo {
|
|
name: name.to_string(),
|
|
data_type: data_type.to_string(),
|
|
is_nullable: true,
|
|
column_default: None,
|
|
is_primary_key: false,
|
|
is_unique: false,
|
|
extra: None,
|
|
comment: None,
|
|
numeric_precision: None,
|
|
numeric_scale: None,
|
|
character_maximum_length: None,
|
|
enum_values: None,
|
|
character_set: None,
|
|
collation: None,
|
|
}
|
|
}
|
|
|
|
// ========================================================================
|
|
// 9.2: Template Engine Tests
|
|
// ========================================================================
|
|
|
|
#[test]
|
|
fn template_engine_creation() {
|
|
let engine = ScriptTemplateEngine::new();
|
|
assert!(engine.env.get_template("schema_sync").is_ok());
|
|
assert!(engine.env.get_template("joint_orchestration").is_ok());
|
|
assert!(engine.env.get_template("batch").is_ok());
|
|
}
|
|
|
|
#[test]
|
|
fn render_schema_sync_template() {
|
|
let engine = ScriptTemplateEngine::new();
|
|
let result = engine.render_schema_sync(
|
|
"CREATE TABLE users (id INT);",
|
|
Some("DROP TABLE IF EXISTS users;"),
|
|
"mysql",
|
|
"mysql",
|
|
"mysql",
|
|
"if_not_exists",
|
|
None,
|
|
"Schema sync for test",
|
|
&[],
|
|
);
|
|
assert!(result.is_ok());
|
|
let output = result.unwrap();
|
|
assert!(output.contains("CREATE TABLE users (id INT);"));
|
|
assert!(output.contains("DROP TABLE IF EXISTS users;"));
|
|
assert!(output.contains("Schema sync for test"));
|
|
}
|
|
|
|
#[test]
|
|
fn render_joint_script_template() {
|
|
let engine = ScriptTemplateEngine::new();
|
|
let plan = JointCorrectionPlan {
|
|
strategy: CorrectionStrategy::StructureFirst,
|
|
steps: vec![CorrectionStep {
|
|
step_type: CorrectionStepType::SchemaCreate,
|
|
table_name: Some("users".to_string()),
|
|
sql: "CREATE TABLE users (id INT);".to_string(),
|
|
rollback_sql: Some("DROP TABLE users;".to_string()),
|
|
description: "Create users table".to_string(),
|
|
risk_level: CorrectionRiskLevel::Safe,
|
|
depends_on: vec![],
|
|
}],
|
|
schema_diff_count: 1,
|
|
data_diff_count: 0,
|
|
total_estimated_duration_secs: None,
|
|
rollback_steps: vec![],
|
|
};
|
|
|
|
let result = engine.render_joint_script(&plan, true, &[]);
|
|
assert!(result.is_ok());
|
|
let output = result.unwrap();
|
|
assert!(output.contains("StructureFirst"));
|
|
assert!(output.contains("CREATE TABLE users"));
|
|
assert!(output.contains("Step 1"));
|
|
}
|
|
|
|
#[test]
|
|
fn render_batch_template() {
|
|
let engine = ScriptTemplateEngine::new();
|
|
let result = engine.render_batch(
|
|
1,
|
|
3,
|
|
"Batch 1: Create tables",
|
|
"CREATE TABLE a (id INT);",
|
|
Some("DROP TABLE a;"),
|
|
Some(100),
|
|
&[],
|
|
);
|
|
assert!(result.is_ok());
|
|
let output = result.unwrap();
|
|
assert!(output.contains("Batch 1 / 3"));
|
|
assert!(output.contains("CREATE TABLE a"));
|
|
assert!(output.contains("DROP TABLE a"));
|
|
assert!(output.contains("CHECKPOINT: batch_1_complete"));
|
|
}
|
|
|
|
// ========================================================================
|
|
// 9.3: Idempotent Strategy Tests
|
|
// ========================================================================
|
|
|
|
#[test]
|
|
fn select_strategy_if_not_exists() {
|
|
let desc = DialectCapabilityDescriptor { flags: CAP_IF_NOT_EXISTS | CAP_CREATE_TABLE, ..Default::default() };
|
|
let strategy = select_strategy(Some(DialectKind::Mysql), Some(&desc));
|
|
assert_eq!(strategy, IdempotentStrategy::IfNotExists);
|
|
}
|
|
|
|
#[test]
|
|
fn select_strategy_create_or_replace() {
|
|
let desc =
|
|
DialectCapabilityDescriptor { flags: CAP_CREATE_OR_REPLACE | CAP_CREATE_TABLE, ..Default::default() };
|
|
let strategy = select_strategy(Some(DialectKind::Postgres), Some(&desc));
|
|
assert_eq!(strategy, IdempotentStrategy::CreateOrReplace);
|
|
}
|
|
|
|
#[test]
|
|
fn select_strategy_conditional() {
|
|
let desc = DialectCapabilityDescriptor { flags: 0, ..Default::default() };
|
|
let strategy = select_strategy(Some(DialectKind::Sqlite), Some(&desc));
|
|
assert_eq!(strategy, IdempotentStrategy::ConditionalCheck);
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_create_table_if_not_exists_mysql() {
|
|
let sql = "CREATE TABLE users (id INT PRIMARY KEY);";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Mysql, IdempotentStrategy::IfNotExists);
|
|
assert!(result.contains("CREATE TABLE IF NOT EXISTS"), "Got: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_create_table_if_not_exists_sqlite() {
|
|
let sql = "CREATE TABLE users (id INT);";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Sqlite, IdempotentStrategy::IfNotExists);
|
|
assert!(result.contains("CREATE TABLE IF NOT EXISTS"), "Got: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_create_table_if_not_exists_clickhouse() {
|
|
let sql = "CREATE TABLE events (ts DateTime);";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::ClickHouse, IdempotentStrategy::IfNotExists);
|
|
assert!(result.contains("CREATE TABLE IF NOT EXISTS"), "Got: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_drop_table_if_exists() {
|
|
let sql = "DROP TABLE old_users;";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Mysql, IdempotentStrategy::IfNotExists);
|
|
assert!(result.contains("DROP TABLE IF EXISTS"), "Got: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_drop_index_if_exists() {
|
|
let sql = "DROP INDEX idx_email;";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Postgres, IdempotentStrategy::IfNotExists);
|
|
assert!(result.contains("DROP INDEX IF EXISTS"), "Got: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_create_index_if_not_exists_postgres() {
|
|
let sql = "CREATE INDEX idx_name ON users (name);";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Postgres, IdempotentStrategy::IfNotExists);
|
|
assert!(result.contains("CREATE INDEX IF NOT EXISTS"), "Got: {result}");
|
|
assert!(!result.contains("CREATE IF NOT EXISTS INDEX"), "wrong IF NOT EXISTS placement: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_create_unique_index_if_not_exists_postgres() {
|
|
let sql = "CREATE UNIQUE INDEX idx_email ON users (email);";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Postgres, IdempotentStrategy::IfNotExists);
|
|
assert!(result.contains("CREATE UNIQUE INDEX IF NOT EXISTS"), "Got: {result}");
|
|
assert!(!result.contains("CREATE IF NOT EXISTS"), "wrong IF NOT EXISTS placement: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_drop_sequence_if_exists() {
|
|
let sql = "DROP SEQUENCE user_seq;";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Postgres, IdempotentStrategy::IfNotExists);
|
|
assert!(result.contains("DROP SEQUENCE IF EXISTS"), "Got: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_drop_function_if_exists() {
|
|
let sql = "DROP FUNCTION calculate_total;";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Postgres, IdempotentStrategy::IfNotExists);
|
|
assert!(result.contains("DROP FUNCTION IF EXISTS"), "Got: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_skip_comments() {
|
|
let sql = "-- This is a comment\nCREATE TABLE users (id INT);";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Mysql, IdempotentStrategy::IfNotExists);
|
|
assert!(result.contains("-- This is a comment"));
|
|
assert!(result.contains("IF NOT EXISTS"));
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_no_wrap_alter_table() {
|
|
let sql = "ALTER TABLE users ADD COLUMN age INT;";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Mysql, IdempotentStrategy::IfNotExists);
|
|
assert!(!result.contains("IF NOT EXISTS"), "ALTER TABLE should not be wrapped: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_already_if_not_exists() {
|
|
let sql = "CREATE TABLE IF NOT EXISTS users (id INT);";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Mysql, IdempotentStrategy::IfNotExists);
|
|
assert_eq!(result.matches("IF NOT EXISTS").count(), 1, "Should not double-wrap: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_already_if_exists_drop() {
|
|
let sql = "DROP TABLE IF EXISTS users;";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Mysql, IdempotentStrategy::IfNotExists);
|
|
assert_eq!(result.matches("IF EXISTS").count(), 1, "Should not double-wrap: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_create_or_replace_view() {
|
|
let sql = "CREATE VIEW user_view AS SELECT * FROM users;";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Postgres, IdempotentStrategy::CreateOrReplace);
|
|
assert!(result.contains("CREATE OR REPLACE"), "Got: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_create_or_replace_function() {
|
|
let sql = "CREATE FUNCTION add(a INT, b INT) RETURNS INT AS $$ BEGIN RETURN a + b; END; $$ LANGUAGE plpgsql;";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Postgres, IdempotentStrategy::CreateOrReplace);
|
|
assert!(result.contains("CREATE OR REPLACE"), "Got: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_create_or_replace_not_for_table() {
|
|
let sql = "CREATE TABLE users (id INT);";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Postgres, IdempotentStrategy::CreateOrReplace);
|
|
assert!(!result.contains("OR REPLACE"), "Tables should not get OR REPLACE: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_none_passthrough() {
|
|
let sql = "CREATE TABLE users (id INT);";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Mysql, IdempotentStrategy::None);
|
|
assert_eq!(result, sql);
|
|
}
|
|
|
|
#[test]
|
|
fn split_sql_respects_escaped_single_quotes() {
|
|
let sql = "INSERT INTO t VALUES ('it''s ok'); CREATE TABLE u (id INT);";
|
|
let parts = split_sql_statements(sql);
|
|
assert_eq!(parts.len(), 2, "parts={parts:?}");
|
|
assert!(parts[0].contains("it''s ok"), "first={:?}", parts[0]);
|
|
assert!(parts[1].to_uppercase().contains("CREATE TABLE"), "second={:?}", parts[1]);
|
|
}
|
|
|
|
#[test]
|
|
fn conditional_check_does_not_emit_unconditional_ddl_after_select() {
|
|
let sql = "CREATE TABLE users (id INT);";
|
|
let mysql = apply_idempotent_strategy(sql, DatabaseType::Mysql, IdempotentStrategy::ConditionalCheck);
|
|
assert!(mysql.contains("CREATE TABLE IF NOT EXISTS"), "mysql={mysql}");
|
|
assert!(!mysql.contains("THEN 'EXECUTE:"), "must not use inert SELECT CASE: {mysql}");
|
|
|
|
let pg = apply_idempotent_strategy(sql, DatabaseType::Postgres, IdempotentStrategy::ConditionalCheck);
|
|
assert!(pg.contains("DO $dbx_idempotent$") || pg.contains("IF NOT EXISTS"), "pg={pg}");
|
|
assert!(!pg.contains("THEN 'EXECUTE:"), "must not use inert SELECT CASE: {pg}");
|
|
// Body is inside EXECUTE dollar-quote, not as a bare follow-up statement.
|
|
let after_do = pg.find("DO $dbx_idempotent$").map(|i| &pg[i..]).unwrap_or(&pg);
|
|
assert!(!after_do.contains("\nCREATE TABLE users"), "bare CREATE must not follow check: {pg}");
|
|
|
|
let oracle = apply_idempotent_strategy(sql, DatabaseType::Oracle, IdempotentStrategy::ConditionalCheck);
|
|
assert!(oracle.contains("EXECUTE IMMEDIATE") || oracle.contains("BEGIN"), "oracle={oracle}");
|
|
assert!(!oracle.contains("THEN 'EXECUTE:"), "must not use inert SELECT CASE: {oracle}");
|
|
}
|
|
|
|
#[test]
|
|
fn conditional_check_postgres_matches_schema_and_table_separately() {
|
|
let qualified = apply_idempotent_strategy(
|
|
"CREATE TABLE sales.users (id INT);",
|
|
DatabaseType::Postgres,
|
|
IdempotentStrategy::ConditionalCheck,
|
|
);
|
|
assert!(qualified.contains("table_schema = 'sales'"), "qualified={qualified}");
|
|
assert!(qualified.contains("table_name = 'users'"), "qualified={qualified}");
|
|
assert!(!qualified.contains("table_name = 'sales.users'"), "qualified={qualified}");
|
|
|
|
let unqualified = apply_idempotent_strategy(
|
|
"CREATE TABLE users (id INT);",
|
|
DatabaseType::Postgres,
|
|
IdempotentStrategy::ConditionalCheck,
|
|
);
|
|
assert!(unqualified.contains("table_schema = current_schema()"), "unqualified={unqualified}");
|
|
|
|
let quoted = apply_idempotent_strategy(
|
|
"CREATE TABLE \"Sales\".\"Users\" (id INT);",
|
|
DatabaseType::Postgres,
|
|
IdempotentStrategy::ConditionalCheck,
|
|
);
|
|
assert!(quoted.contains("table_schema = 'Sales'"), "quoted={quoted}");
|
|
assert!(quoted.contains("table_name = 'Users'"), "quoted={quoted}");
|
|
}
|
|
|
|
#[test]
|
|
fn conditional_check_object_exists_path_uses_if_exists_when_available() {
|
|
let sql = "DROP TABLE users;";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Postgres, IdempotentStrategy::ConditionalCheck);
|
|
assert!(result.contains("DROP TABLE IF EXISTS") || result.contains("Conditional:"), "Got: {result}");
|
|
assert!(!result.contains("THEN 'EXECUTE:"), "{result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_empty_string() {
|
|
let result = apply_idempotent_strategy("", DatabaseType::Mysql, IdempotentStrategy::IfNotExists);
|
|
assert_eq!(result, "");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_conditional_check_drop() {
|
|
let sql = "DROP TABLE old_data;";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Sqlite, IdempotentStrategy::ConditionalCheck);
|
|
// Prefer real IF EXISTS over an inert comment+DDL pair.
|
|
assert!(result.contains("DROP TABLE IF EXISTS") || result.contains("Conditional"), "Got: {result}");
|
|
assert!(result.contains("old_data"), "Got: {result}");
|
|
assert!(!result.contains("THEN 'EXECUTE:"), "{result}");
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_create_or_replace_already_has_or_replace() {
|
|
let sql = "CREATE OR REPLACE VIEW v AS SELECT 1;";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Postgres, IdempotentStrategy::CreateOrReplace);
|
|
assert_eq!(result.matches("OR REPLACE").count(), 1, "Should not double-wrap: {result}");
|
|
}
|
|
|
|
// ========================================================================
|
|
// 9.4: Rollback Script Generator Tests
|
|
// ========================================================================
|
|
|
|
#[test]
|
|
fn generate_rollback_script_from_graph() {
|
|
let table_diff = TableDiff {
|
|
diff_type: "added".to_string(),
|
|
object_type: Some("table".to_string()),
|
|
name: "users".to_string(),
|
|
columns: Some(vec![ColumnDiff {
|
|
diff_type: "added".to_string(),
|
|
name: "id".to_string(),
|
|
source: Some(make_column_info("id", "INT")),
|
|
target: None,
|
|
changes: vec![],
|
|
add_position: None,
|
|
}]),
|
|
indexes: Some(vec![]),
|
|
foreign_keys: Some(vec![]),
|
|
triggers: Some(vec![]),
|
|
ddl: Some("CREATE TABLE users (id INT);".to_string()),
|
|
target_ddl: None,
|
|
source_table_comment: None,
|
|
target_table_comment: None,
|
|
sync_sql: None,
|
|
};
|
|
|
|
let mut graph = RollbackGraph {
|
|
forward_nodes: vec![DiffNode {
|
|
table_diff: table_diff.clone(),
|
|
direction: crate::schema_diff::DiffDirection::Forward,
|
|
dependency_order: 0,
|
|
rename_source: None,
|
|
rename_target: None,
|
|
rename_score: None,
|
|
}],
|
|
rollback_nodes: vec![DiffNode {
|
|
table_diff: TableDiff {
|
|
diff_type: "removed".to_string(),
|
|
object_type: Some("table".to_string()),
|
|
name: "users".to_string(),
|
|
columns: Some(vec![ColumnDiff {
|
|
diff_type: "removed".to_string(),
|
|
name: "id".to_string(),
|
|
source: Some(make_column_info("id", "INT")),
|
|
target: None,
|
|
changes: vec![],
|
|
add_position: None,
|
|
}]),
|
|
indexes: Some(vec![]),
|
|
foreign_keys: Some(vec![]),
|
|
triggers: Some(vec![]),
|
|
ddl: None,
|
|
target_ddl: Some("CREATE TABLE users (id INT);".to_string()),
|
|
source_table_comment: None,
|
|
target_table_comment: None,
|
|
sync_sql: None,
|
|
},
|
|
direction: crate::schema_diff::DiffDirection::Rollback,
|
|
dependency_order: 0,
|
|
rename_source: None,
|
|
rename_target: None,
|
|
rename_score: None,
|
|
}],
|
|
is_consistent: true,
|
|
consistency_issues: vec![],
|
|
};
|
|
graph.is_consistent = true;
|
|
|
|
let options = RollbackScriptOptions {
|
|
db_type: DatabaseType::Postgres,
|
|
target_schema: Some("public".to_string()),
|
|
cascade_delete: false,
|
|
include_header: true,
|
|
include_safety_checks: true,
|
|
idempotent_strategy: IdempotentStrategy::IfNotExists,
|
|
};
|
|
|
|
let script = generate_rollback_script(&graph, &options);
|
|
assert!(script.contains("ROLLBACK SCRIPT"));
|
|
assert!(script.contains("IF EXISTS"), "Got: {script}");
|
|
}
|
|
|
|
#[test]
|
|
fn generate_rollback_script_with_inconsistency_warning() {
|
|
let graph = RollbackGraph {
|
|
forward_nodes: vec![],
|
|
rollback_nodes: vec![],
|
|
is_consistent: false,
|
|
consistency_issues: vec!["Missing rollback for added table 'users'".to_string()],
|
|
};
|
|
|
|
let options = RollbackScriptOptions::default();
|
|
let script = generate_rollback_script(&graph, &options);
|
|
assert!(script.contains("WARNING"));
|
|
assert!(script.contains("Missing rollback"));
|
|
}
|
|
|
|
#[test]
|
|
fn generate_rollback_script_empty_graph() {
|
|
let graph = RollbackGraph {
|
|
forward_nodes: vec![],
|
|
rollback_nodes: vec![],
|
|
is_consistent: true,
|
|
consistency_issues: vec![],
|
|
};
|
|
|
|
let options = RollbackScriptOptions::default();
|
|
let script = generate_rollback_script(&graph, &options);
|
|
assert!(script.contains("ROLLBACK SCRIPT"));
|
|
}
|
|
|
|
#[test]
|
|
fn rollback_script_respects_no_header_option() {
|
|
let graph = RollbackGraph {
|
|
forward_nodes: vec![],
|
|
rollback_nodes: vec![],
|
|
is_consistent: true,
|
|
consistency_issues: vec![],
|
|
};
|
|
|
|
let options =
|
|
RollbackScriptOptions { include_header: false, include_safety_checks: false, ..Default::default() };
|
|
|
|
let script = generate_rollback_script(&graph, &options);
|
|
assert!(!script.contains("ROLLBACK SCRIPT"));
|
|
}
|
|
|
|
// ========================================================================
|
|
// 9.5: Joint Script Tests
|
|
// ========================================================================
|
|
|
|
#[test]
|
|
fn generate_joint_script_structure_first() {
|
|
let schema_diff = make_simple_schema_diff();
|
|
let options = JointScriptOptions {
|
|
strategy: JointScriptStrategy::StructureFirst,
|
|
include_checkpoints: true,
|
|
include_rollback: true,
|
|
db_type: DatabaseType::Mysql,
|
|
..Default::default()
|
|
};
|
|
|
|
let script = generate_joint_script(&schema_diff, None, &options);
|
|
assert!(script.contains("StructureFirst"), "Got: {script}");
|
|
assert!(script.contains("PHASE 1: Structure"));
|
|
assert!(script.contains("CHECKPOINT"));
|
|
}
|
|
|
|
#[test]
|
|
fn generate_joint_script_data_first() {
|
|
let schema_diff = make_simple_schema_diff();
|
|
let options = JointScriptOptions {
|
|
strategy: JointScriptStrategy::DataFirst,
|
|
db_type: DatabaseType::Postgres,
|
|
..Default::default()
|
|
};
|
|
|
|
let script = generate_joint_script(&schema_diff, None, &options);
|
|
assert!(script.contains("PHASE 1: Data"));
|
|
assert!(script.contains("PHASE 2: Structure"));
|
|
}
|
|
|
|
#[test]
|
|
fn generate_joint_script_shadow_table() {
|
|
let schema_diff = SchemaDiffPreparation {
|
|
diffs: vec![TableDiff {
|
|
diff_type: "modified".to_string(),
|
|
object_type: Some("table".to_string()),
|
|
name: "users".to_string(),
|
|
columns: Some(vec![]),
|
|
indexes: Some(vec![]),
|
|
foreign_keys: Some(vec![]),
|
|
triggers: Some(vec![]),
|
|
ddl: Some("CREATE TABLE users (id INT, name VARCHAR(100));".to_string()),
|
|
target_ddl: Some("CREATE TABLE users (id INT);".to_string()),
|
|
source_table_comment: None,
|
|
target_table_comment: None,
|
|
sync_sql: None,
|
|
}],
|
|
sync_sql: "ALTER TABLE users ADD COLUMN name VARCHAR(100);".to_string(),
|
|
..Default::default()
|
|
};
|
|
|
|
let options = JointScriptOptions {
|
|
strategy: JointScriptStrategy::ShadowTable,
|
|
db_type: DatabaseType::Postgres,
|
|
..Default::default()
|
|
};
|
|
|
|
let script = generate_joint_script(&schema_diff, None, &options);
|
|
assert!(script.contains("Shadow Table"));
|
|
assert!(script.contains("_shadow_users"), "Got: {script}");
|
|
assert!(script.contains("RENAME TO"), "Got: {script}");
|
|
}
|
|
|
|
#[test]
|
|
fn generate_joint_script_no_rollback() {
|
|
let schema_diff = make_simple_schema_diff();
|
|
let options = JointScriptOptions {
|
|
strategy: JointScriptStrategy::StructureFirst,
|
|
include_rollback: false,
|
|
db_type: DatabaseType::Mysql,
|
|
..Default::default()
|
|
};
|
|
|
|
let script = generate_joint_script(&schema_diff, None, &options);
|
|
assert!(!script.contains("ROLLBACK SECTION"));
|
|
}
|
|
|
|
// Helper: minimal SchemaDiffPreparation for tests
|
|
fn make_simple_schema_diff() -> SchemaDiffPreparation {
|
|
SchemaDiffPreparation {
|
|
diffs: vec![TableDiff {
|
|
diff_type: "added".to_string(),
|
|
object_type: Some("table".to_string()),
|
|
name: "test_table".to_string(),
|
|
columns: Some(vec![]),
|
|
indexes: Some(vec![]),
|
|
foreign_keys: Some(vec![]),
|
|
triggers: Some(vec![]),
|
|
ddl: Some("CREATE TABLE test_table (id INT PRIMARY KEY);".to_string()),
|
|
target_ddl: None,
|
|
source_table_comment: None,
|
|
target_table_comment: None,
|
|
sync_sql: Some("CREATE TABLE test_table (id INT PRIMARY KEY);".to_string()),
|
|
}],
|
|
sync_sql: "CREATE TABLE test_table (id INT PRIMARY KEY);".to_string(),
|
|
rollback_sync_sql: Some("DROP TABLE IF EXISTS test_table;".to_string()),
|
|
..Default::default()
|
|
}
|
|
}
|
|
|
|
impl Default for SchemaDiffPreparation {
|
|
fn default() -> Self {
|
|
Self {
|
|
diffs: vec![],
|
|
function_diffs: vec![],
|
|
sequence_diffs: vec![],
|
|
rule_diffs: vec![],
|
|
owner_diffs: vec![],
|
|
sync_sql: String::new(),
|
|
rollback_sync_sql: None,
|
|
rollback_completeness: crate::schema_diff::RollbackCompleteness::Complete,
|
|
missing_rollback_objects: vec![],
|
|
rename_candidates: vec![],
|
|
rollback_graph: None,
|
|
compatibility_warnings: vec![],
|
|
permission_diffs: vec![],
|
|
permission_sync_sql: None,
|
|
dependency_graph: None,
|
|
}
|
|
}
|
|
}
|
|
|
|
// ========================================================================
|
|
// 9.6: Batch Controller Tests
|
|
// ========================================================================
|
|
|
|
#[test]
|
|
fn build_batches_single_batch() {
|
|
let config = BatchConfig { max_statements_per_batch: 10, ..Default::default() };
|
|
let controller = BatchController::new(config);
|
|
let sql = "CREATE TABLE a (id INT);\nCREATE TABLE b (id INT);";
|
|
let table_infos = vec![make_table_info("a"), make_table_info("b")];
|
|
|
|
let batches = controller.build_batches(sql, None, &table_infos);
|
|
assert_eq!(batches.len(), 1, "Should have 1 batch for 2 statements with batch_size=10");
|
|
assert_eq!(batches[0].batch_id, 1);
|
|
assert!(batches[0].sql.contains("Batch 1"));
|
|
}
|
|
|
|
#[test]
|
|
fn build_batches_multiple_batches() {
|
|
let config = BatchConfig { max_statements_per_batch: 2, ..Default::default() };
|
|
let controller = BatchController::new(config);
|
|
|
|
let mut statements = String::new();
|
|
for i in 0..5 {
|
|
statements.push_str(&format!("CREATE TABLE t{i} (id INT);\n"));
|
|
}
|
|
|
|
let batches = controller.build_batches(&statements, None, &[]);
|
|
assert_eq!(batches.len(), 3, "5 statements with batch_size=2 should give 3 batches");
|
|
assert_eq!(batches[0].batch_id, 1);
|
|
assert_eq!(batches[1].batch_id, 2);
|
|
assert_eq!(batches[2].batch_id, 3);
|
|
}
|
|
|
|
#[test]
|
|
fn build_batches_empty_sql() {
|
|
let config = BatchConfig::default();
|
|
let controller = BatchController::new(config);
|
|
let batches = controller.build_batches("", None, &[]);
|
|
assert!(batches.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn build_batches_with_table_names() {
|
|
let config = BatchConfig { max_statements_per_batch: 1, ..Default::default() };
|
|
let controller = BatchController::new(config);
|
|
let sql =
|
|
"CREATE TABLE users (id INT);\nALTER TABLE users ADD COLUMN name VARCHAR(100);\nDROP TABLE old_table;";
|
|
|
|
let batches = controller.build_batches(sql, None, &[]);
|
|
assert_eq!(batches.len(), 3);
|
|
|
|
assert!(
|
|
batches[0].table_names.contains(&"users".to_string()),
|
|
"Batch 0 table_names: {:?}",
|
|
batches[0].table_names
|
|
);
|
|
assert!(
|
|
batches[1].table_names.contains(&"users".to_string()),
|
|
"Batch 1 table_names: {:?}",
|
|
batches[1].table_names
|
|
);
|
|
assert!(
|
|
batches[2].table_names.contains(&"old_table".to_string()),
|
|
"Batch 2 table_names: {:?}",
|
|
batches[2].table_names
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn batch_state_checkpoint_key() {
|
|
let config = BatchConfig { checkpoint_key_prefix: "my_sync".to_string(), ..Default::default() };
|
|
assert_eq!(BatchState::checkpoint_key(&config), "my_sync_state");
|
|
}
|
|
|
|
#[test]
|
|
fn batch_state_new() {
|
|
let state = BatchState::new(5);
|
|
assert_eq!(state.total_batches, 5);
|
|
assert_eq!(state.status, "created");
|
|
assert!(state.completed_batches.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn batch_controller_get_next_batch() {
|
|
let config = BatchConfig { max_statements_per_batch: 1, ..Default::default() };
|
|
let controller = BatchController::new(config);
|
|
|
|
let sql = "CREATE TABLE a (id INT);\nCREATE TABLE b (id INT);\nCREATE TABLE c (id INT);";
|
|
let batches = controller.build_batches(sql, None, &[]);
|
|
|
|
let mut state = BatchState::new(batches.len());
|
|
|
|
let next = controller.get_next_batch(&state, &batches);
|
|
assert!(next.is_some());
|
|
assert_eq!(next.unwrap().batch_id, 1);
|
|
|
|
controller.mark_batch_complete(&mut state, 1);
|
|
let next = controller.get_next_batch(&state, &batches);
|
|
assert!(next.is_some());
|
|
assert_eq!(next.unwrap().batch_id, 2);
|
|
}
|
|
|
|
#[test]
|
|
fn batch_controller_mark_all_complete() {
|
|
let config = BatchConfig::default();
|
|
let controller = BatchController::new(config);
|
|
let mut state = BatchState::new(2);
|
|
|
|
assert_eq!(state.status, "created");
|
|
controller.mark_batch_complete(&mut state, 1);
|
|
controller.mark_batch_complete(&mut state, 2);
|
|
assert_eq!(state.status, "completed");
|
|
}
|
|
|
|
#[test]
|
|
fn batch_controller_no_duplicate_completion() {
|
|
let config = BatchConfig::default();
|
|
let controller = BatchController::new(config);
|
|
let mut state = BatchState::new(2);
|
|
|
|
controller.mark_batch_complete(&mut state, 1);
|
|
controller.mark_batch_complete(&mut state, 1);
|
|
assert_eq!(state.completed_batches.len(), 1, "Should not duplicate");
|
|
}
|
|
|
|
// ========================================================================
|
|
// Enhanced Functions Tests
|
|
// ========================================================================
|
|
|
|
#[test]
|
|
fn generate_enhanced_sync_sql_with_idempotent() {
|
|
let schema_diff =
|
|
SchemaDiffPreparation { sync_sql: "CREATE TABLE test (id INT);".to_string(), ..Default::default() };
|
|
|
|
let result =
|
|
generate_enhanced_sync_sql(&schema_diff, DatabaseType::Mysql, None, IdempotentStrategy::IfNotExists);
|
|
assert!(result.contains("IF NOT EXISTS"), "Got: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn generate_enhanced_rollback_from_graph() {
|
|
let table_diff = TableDiff {
|
|
diff_type: "added".to_string(),
|
|
object_type: Some("table".to_string()),
|
|
name: "users".to_string(),
|
|
columns: Some(vec![]),
|
|
indexes: Some(vec![]),
|
|
foreign_keys: Some(vec![]),
|
|
triggers: Some(vec![]),
|
|
ddl: Some("CREATE TABLE users (id INT);".to_string()),
|
|
target_ddl: None,
|
|
source_table_comment: None,
|
|
target_table_comment: None,
|
|
sync_sql: None,
|
|
};
|
|
|
|
let graph = RollbackGraph {
|
|
forward_nodes: vec![DiffNode {
|
|
table_diff: table_diff.clone(),
|
|
direction: crate::schema_diff::DiffDirection::Forward,
|
|
dependency_order: 0,
|
|
rename_source: None,
|
|
rename_target: None,
|
|
rename_score: None,
|
|
}],
|
|
rollback_nodes: vec![DiffNode {
|
|
table_diff: TableDiff {
|
|
diff_type: "removed".to_string(),
|
|
object_type: Some("table".to_string()),
|
|
name: "users".to_string(),
|
|
columns: Some(vec![]),
|
|
indexes: Some(vec![]),
|
|
foreign_keys: Some(vec![]),
|
|
triggers: Some(vec![]),
|
|
ddl: None,
|
|
target_ddl: Some("CREATE TABLE users (id INT);".to_string()),
|
|
source_table_comment: None,
|
|
target_table_comment: None,
|
|
sync_sql: None,
|
|
},
|
|
direction: crate::schema_diff::DiffDirection::Rollback,
|
|
dependency_order: 0,
|
|
rename_source: None,
|
|
rename_target: None,
|
|
rename_score: None,
|
|
}],
|
|
is_consistent: true,
|
|
consistency_issues: vec![],
|
|
};
|
|
|
|
let schema_diff = SchemaDiffPreparation {
|
|
sync_sql: "CREATE TABLE users (id INT);".to_string(),
|
|
rollback_sync_sql: Some("DROP TABLE IF EXISTS users;".to_string()),
|
|
rollback_graph: Some(graph),
|
|
..Default::default()
|
|
};
|
|
|
|
let result = generate_enhanced_rollback_sql(
|
|
&schema_diff,
|
|
DatabaseType::Postgres,
|
|
Some("public"),
|
|
false,
|
|
IdempotentStrategy::IfNotExists,
|
|
);
|
|
assert!(result.is_some());
|
|
assert!(result.unwrap().contains("ROLLBACK SCRIPT"));
|
|
}
|
|
|
|
#[test]
|
|
fn generate_complete_script_with_all_sections() {
|
|
let schema_diff = SchemaDiffPreparation {
|
|
sync_sql: "CREATE TABLE test (id INT);".to_string(),
|
|
rollback_sync_sql: Some("DROP TABLE IF EXISTS test;".to_string()),
|
|
permission_sync_sql: Some("GRANT SELECT ON test TO reader;".to_string()),
|
|
..Default::default()
|
|
};
|
|
|
|
let script = generate_complete_script(
|
|
&schema_diff,
|
|
None,
|
|
DatabaseType::Postgres,
|
|
Some("public"),
|
|
IdempotentStrategy::IfNotExists,
|
|
);
|
|
|
|
assert!(script.contains("Structure Changes"));
|
|
assert!(script.contains("Permission Changes"));
|
|
assert!(script.contains("Rollback Script"));
|
|
assert!(script.contains("public"));
|
|
}
|
|
|
|
#[test]
|
|
fn idempotent_mysql_table_contains_if_not_exists_no_double_wrap() {
|
|
let sql = "CREATE TABLE IF NOT EXISTS existing (id INT);";
|
|
let result = apply_idempotent_strategy(sql, DatabaseType::Mysql, IdempotentStrategy::IfNotExists);
|
|
assert_eq!(result.matches("IF NOT EXISTS").count(), 1, "Already contains IF NOT EXISTS: {result}");
|
|
}
|
|
|
|
#[test]
|
|
fn extract_table_name_from_create() {
|
|
assert_eq!(extract_table_name("CREATE TABLE users (id INT)", "TABLE"), "users");
|
|
assert_eq!(extract_table_name("CREATE TABLE `my_table` (id INT)", "TABLE"), "my_table");
|
|
assert_eq!(extract_table_name("DROP TABLE old_data;", "TABLE"), "old_data");
|
|
}
|
|
|
|
#[test]
|
|
fn extract_from_alter_table() {
|
|
assert_eq!(extract_from_alter("ALTER TABLE users ADD COLUMN name VARCHAR(100);"), Some("users".to_string()));
|
|
assert_eq!(
|
|
extract_from_alter("ALTER TABLE \"schema\".\"users\" ADD COLUMN name VARCHAR(100);"),
|
|
Some("schema\".\"users".to_string())
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn split_sql_statements_basic() {
|
|
let sql = "CREATE TABLE a (id INT);\nCREATE TABLE b (id INT);";
|
|
let stmts = split_sql_statements(sql);
|
|
assert_eq!(stmts.len(), 2);
|
|
}
|
|
|
|
#[test]
|
|
fn split_sql_statements_with_string_literal() {
|
|
let sql = "INSERT INTO t VALUES ('hello; world');\nSELECT * FROM t;";
|
|
let stmts = split_sql_statements(sql);
|
|
assert_eq!(stmts.len(), 2);
|
|
assert!(stmts[0].contains("hello; world"));
|
|
}
|
|
|
|
#[test]
|
|
fn split_sql_statements_trailing_without_semicolon() {
|
|
let sql = "CREATE TABLE a (id INT);\nSELECT 1";
|
|
let stmts = split_sql_statements(sql);
|
|
assert_eq!(stmts.len(), 2);
|
|
}
|
|
|
|
#[test]
|
|
fn generate_dml_clean_integration() {
|
|
use crate::schema_diff::ColumnDiff;
|
|
use crate::types::ColumnInfo;
|
|
|
|
let col_diff = ColumnDiff {
|
|
diff_type: "modified".to_string(),
|
|
name: "large_val".to_string(),
|
|
source: Some(ColumnInfo {
|
|
name: "large_val".to_string(),
|
|
data_type: "BIGINT".to_string(),
|
|
is_nullable: true,
|
|
column_default: None,
|
|
is_primary_key: false,
|
|
is_unique: false,
|
|
extra: None,
|
|
comment: None,
|
|
numeric_precision: None,
|
|
numeric_scale: None,
|
|
character_maximum_length: None,
|
|
enum_values: None,
|
|
character_set: None,
|
|
collation: None,
|
|
}),
|
|
target: Some(ColumnInfo {
|
|
name: "large_val".to_string(),
|
|
data_type: "INT".to_string(),
|
|
is_nullable: true,
|
|
column_default: None,
|
|
is_primary_key: false,
|
|
is_unique: false,
|
|
extra: None,
|
|
comment: None,
|
|
numeric_precision: None,
|
|
numeric_scale: None,
|
|
character_maximum_length: None,
|
|
enum_values: None,
|
|
character_set: None,
|
|
collation: None,
|
|
}),
|
|
changes: vec!["data_type BIGINT → INT".to_string()],
|
|
add_position: None,
|
|
};
|
|
|
|
let table_diff = TableDiff {
|
|
diff_type: "modified".to_string(),
|
|
object_type: Some("TABLE".to_string()),
|
|
name: "test_table".to_string(),
|
|
columns: Some(vec![col_diff]),
|
|
..Default::default()
|
|
};
|
|
|
|
let prep = SchemaDiffPreparation { diffs: vec![table_diff], ..Default::default() };
|
|
|
|
// Seed a pair-specific rule so the test does not depend on CWD for YAML files.
|
|
{
|
|
use crate::dml_binding::{DmlCleanRule, DmlCleanRuleRegistry};
|
|
if let Ok(mut registry) = DmlCleanRuleRegistry::global().write() {
|
|
registry.rules = vec![DmlCleanRule {
|
|
name: "bigint_to_int".to_string(),
|
|
source_type: "BIGINT".to_string(),
|
|
target_type: "INT".to_string(),
|
|
max_fidelity: 1.0,
|
|
pre_transform_sql: Some(
|
|
"UPDATE {table} SET {column} = NULL WHERE {column} > 2147483647".to_string(),
|
|
),
|
|
description: None,
|
|
}];
|
|
}
|
|
}
|
|
|
|
let sql = generate_all_dml_clean_sql(&prep, 0.8);
|
|
assert!(!sql.is_empty(), "DML should be generated for BIGINT→INT");
|
|
assert!(sql[0].contains("2147483647"));
|
|
}
|
|
}
|