use crate::abstract_layer::DbType;
use crate::abstract_layer::common::{Database, Transaction};
#[cfg_attr(not(feature = "postgresql"), allow(unused_imports))]
use crate::migration::split_qualified_table_name;
use crate::migration::{
execute_steps, execute_steps_nontransactional, is_schema_rebuild_error, table_name,
ExpectedIndexDef, MigrationPlan, MigrationStep, TableMigration,
};
use crate::model::WritableModel;
use std::collections::BTreeSet;
pub type TablePlan = MigrationPlan;
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TableDiagnosis {
Ready,
Migratable(TablePlan),
NeedsRebuild(RebuildCause),
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RebuildCause {
pub table: String,
pub reason: String,
}
impl RebuildCause {
pub(crate) fn new(table: impl Into<String>, reason: impl Into<String>) -> Self {
Self {
table: table.into(),
reason: reason.into(),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ExtraPolicy {
Keep,
Drop,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RebuildPolicy {
Refuse,
Allow,
}
#[derive(Debug, Clone)]
pub struct ApplyOptions {
pub extra_columns: ExtraPolicy,
pub rebuild: RebuildPolicy,
pub index_concurrently: bool,
pub extensions: Vec<String>,
}
impl Default for ApplyOptions {
fn default() -> Self {
Self {
extra_columns: ExtraPolicy::Keep,
rebuild: RebuildPolicy::Refuse,
index_concurrently: true,
extensions: Vec::new(),
}
}
}
impl ApplyOptions {
pub(crate) fn strict() -> Self {
Self {
index_concurrently: false,
..Self::default()
}
}
pub(crate) fn permissive() -> Self {
Self {
extra_columns: ExtraPolicy::Drop,
rebuild: RebuildPolicy::Allow,
index_concurrently: false,
extensions: Vec::new(),
}
}
}
#[derive(Debug)]
pub struct TableApplyOutcome {
pub diagnosis: TableDiagnosis,
pub created_table: bool,
pub executed: Vec<MigrationStep>,
}
fn collapse_whitespace(value: &str) -> String {
value.split_whitespace().collect::<Vec<_>>().join(" ")
}
fn trim_redundant_parens(value: &str) -> String {
let mut current = collapse_whitespace(value);
for _ in 0..64 {
let trimmed = current.trim();
if !(trimmed.starts_with('(') && trimmed.ends_with(')')) || trimmed.len() < 2 {
return trimmed.to_string();
}
let mut depth = 0usize;
let mut closes_at_end = false;
let mut inner_top_comma = false;
for (idx, ch) in trimmed.char_indices() {
match ch {
'(' => depth += 1,
')' => {
depth = depth.saturating_sub(1);
if depth == 0 {
closes_at_end = idx + ch.len_utf8() == trimmed.len();
}
}
',' if depth == 1 => inner_top_comma = true,
_ => {}
}
}
if !closes_at_end || inner_top_comma || depth != 0 {
return trimmed.to_string();
}
let inner = trimmed[1..trimmed.len() - 1].trim();
if inner.is_empty() {
return trimmed.to_string();
}
current = inner.to_string();
}
current
}
pub(crate) fn norm_sql_fragment(value: &str) -> String {
let no_quotes = value.replace(['"', '`'], "");
trim_redundant_parens(&no_quotes)
}
#[cfg_attr(not(feature = "postgresql"), allow(dead_code))]
fn strip_type_casts(value: &str) -> String {
let mut out = String::with_capacity(value.len());
let mut chars = value.chars().peekable();
let mut in_string = false;
while let Some(ch) = chars.next() {
if ch == '\'' {
in_string = !in_string;
out.push(ch);
continue;
}
if !in_string && ch == ':' && chars.peek() == Some(&':') {
chars.next();
while let Some(&next) = chars.peek() {
if next.is_ascii_alphanumeric() || next == '_' || next == '[' || next == ']' {
chars.next();
} else {
break;
}
}
continue;
}
out.push(ch);
}
out
}
#[cfg_attr(not(feature = "postgresql"), allow(dead_code))]
fn normalize_parens(input: &str) -> String {
let chars: Vec<char> = collapse_whitespace(input).chars().collect();
let mut out = String::with_capacity(chars.len());
let mut index = 0usize;
while index < chars.len() {
let ch = chars[index];
if ch == '\'' {
out.push(ch);
index += 1;
while index < chars.len() {
out.push(chars[index]);
if chars[index] == '\'' {
index += 1;
break;
}
index += 1;
}
continue;
}
if ch == '(' {
let mut depth = 0usize;
let mut close = None;
for (k, &matched) in chars.iter().enumerate().skip(index) {
match matched {
'(' => depth += 1,
')' => {
depth -= 1;
if depth == 0 {
close = Some(k);
break;
}
}
_ => {}
}
}
let Some(close) = close else {
out.extend(chars[index..].iter());
break;
};
let inner: String = chars[index + 1..close].iter().collect();
let prev = if index == 0 {
' '
} else {
chars[index - 1]
};
let prev_is_ident =
prev.is_ascii_alphanumeric() || prev == '_' || prev == ')' || prev == ']';
let has_top_level_comma = {
let mut depth = 0usize;
let mut comma = false;
for inner_ch in inner.chars() {
match inner_ch {
'(' => depth += 1,
')' => depth = depth.saturating_sub(1),
',' if depth == 0 => {
comma = true;
break;
}
_ => {}
}
}
comma
};
let inner_normalized = normalize_parens(&inner);
if !prev_is_ident && !has_top_level_comma {
out.push_str(&inner_normalized);
} else {
out.push('(');
out.push_str(&inner_normalized);
out.push(')');
}
index = close + 1;
continue;
}
out.push(ch);
index += 1;
}
out
}
#[cfg_attr(not(feature = "postgresql"), allow(dead_code))]
pub(crate) fn normalize_check_expression(value: &str) -> String {
let fragment = collapse_whitespace(value);
let without_keyword = if fragment.to_ascii_uppercase().starts_with("CHECK ") {
fragment[6..].trim()
} else {
fragment.as_str()
};
let without_cast = strip_type_casts(without_keyword);
normalize_parens(&without_cast)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct PgIndexDefinition {
pub method: String,
pub columns: String,
pub predicate: Option<String>,
}
pub(crate) fn parse_pg_index_definition(definition: &str) -> Option<PgIndexDefinition> {
let upper = definition.to_ascii_uppercase();
let using_pos = upper.find(" USING ")?;
let after_using = &definition[using_pos + " USING ".len()..];
let method = after_using
.split_whitespace()
.next()?
.to_ascii_lowercase();
let rest = &after_using[method.len()..];
let open = rest.find('(')?;
let bytes = rest.as_bytes();
let mut depth = 0usize;
let mut close = None;
for (offset, ch) in bytes[open..].iter().enumerate() {
match ch {
b'(' => depth += 1,
b')' => {
depth -= 1;
if depth == 0 {
close = Some(open + offset);
break;
}
}
_ => {}
}
}
let close = close?;
let columns = rest[open + 1..close].to_string();
let tail = &rest[close + 1..];
let tail_upper = tail.to_ascii_uppercase();
let predicate = tail_upper
.find(" WHERE ")
.map(|pos| tail[pos + " WHERE ".len()..].trim().to_string())
.filter(|value| !value.is_empty());
Some(PgIndexDefinition {
method,
columns,
predicate,
})
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ActualIndexFacts {
pub name: String,
pub columns: Vec<(String, bool)>,
pub unique: bool,
pub valid: bool,
pub method: Option<String>,
pub definition: Option<String>,
pub predicate: Option<String>,
}
impl ActualIndexFacts {
pub(crate) fn from_plain(
name: impl Into<String>,
columns: Vec<(String, bool)>,
unique: bool,
) -> Self {
Self {
name: name.into(),
columns,
unique,
valid: true,
method: None,
definition: None,
predicate: None,
}
}
fn plain_key(&self) -> (bool, Vec<(String, bool)>) {
(self.unique, self.columns.clone())
}
}
pub(crate) fn match_index_semantics(
expected_unique: bool,
expected_method: Option<&str>,
expected_columns: &[(String, bool)],
expected_expression: Option<&str>,
expected_predicate: Option<&str>,
actual: &ActualIndexFacts,
) -> bool {
if !actual.valid {
return false;
}
if expected_unique != actual.unique {
return false;
}
let actual_method = actual
.method
.as_deref()
.unwrap_or("btree");
let expected_method_value = expected_method.unwrap_or("btree");
if !expected_method_value.eq_ignore_ascii_case(actual_method) {
return false;
}
if let Some(expected_expression_value) = expected_expression {
let Some(definition) = &actual.definition else {
return false;
};
let Some(parsed) = parse_pg_index_definition(definition) else {
return false;
};
let actual_columns_norm = norm_sql_fragment(&parsed.columns);
let expected_columns_norm = norm_sql_fragment(expected_expression_value);
if actual_columns_norm != expected_columns_norm {
return false;
}
let actual_predicate = parsed.predicate.as_deref().map(norm_sql_fragment);
let expected_predicate_value = expected_predicate.map(norm_sql_fragment);
return actual_predicate == expected_predicate_value;
}
if actual.predicate.is_some() {
return false;
}
compare_plain_columns(expected_columns, actual)
}
fn compare_plain_columns(expected: &[(String, bool)], actual: &ActualIndexFacts) -> bool {
if let Some(definition) = &actual.definition {
if let Some(parsed) = parse_pg_index_definition(definition) {
let actual_norm = norm_sql_fragment(&parsed.columns);
let expected_text = expected
.iter()
.map(|(name, descending)| {
if *descending {
format!("{name} DESC")
} else {
name.clone()
}
})
.collect::<Vec<_>>()
.join(", ");
return actual_norm == norm_sql_fragment(&expected_text);
}
}
actual.plain_key().1 == expected.to_vec()
}
#[cfg_attr(not(feature = "postgresql"), allow(dead_code))]
pub(crate) fn parse_plain_index_column_list(columns: &str) -> Vec<(String, bool)> {
let mut parts: Vec<String> = Vec::new();
let mut current = String::new();
let mut depth = 0usize;
for ch in columns.chars() {
match ch {
'(' => {
depth += 1;
current.push(ch);
}
')' => {
depth = depth.saturating_sub(1);
current.push(ch);
}
',' if depth == 0 => {
parts.push(std::mem::take(&mut current));
}
_ => current.push(ch),
}
}
parts.push(current);
parts
.iter()
.filter_map(|part| {
let trimmed = part.trim();
if trimmed.is_empty() {
return None;
}
let upper = trimmed.to_ascii_uppercase();
let (name, descending) = if upper.ends_with(" DESC") {
(trimmed[..trimmed.len() - 5].trim(), true)
} else if upper.ends_with(" ASC") {
(trimmed[..trimmed.len() - 4].trim(), false)
} else {
(trimmed, false)
};
Some((norm_sql_fragment(name), descending))
})
.collect()
}
#[cfg_attr(not(feature = "postgresql"), allow(dead_code))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ExpectedCheck {
pub name: String,
pub expr: String,
}
#[cfg_attr(not(feature = "postgresql"), allow(dead_code))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ActualCheckFacts {
pub name: String,
pub definition: String,
}
#[cfg_attr(not(feature = "postgresql"), allow(dead_code))]
pub(crate) fn plan_check_diff(
table_name: &str,
expected: &[ExpectedCheck],
actual: &[ActualCheckFacts],
db_type: DbType,
) -> Vec<MigrationStep> {
let mut steps = Vec::new();
for check in expected {
let expected_norm = normalize_check_expression(&check.expr);
let same_name = actual
.iter()
.find(|candidate| candidate.name == check.name);
match same_name {
Some(existing) => {
let existing_expr = extract_check_expr(&existing.definition);
if normalize_check_expression(&existing_expr) == expected_norm {
continue;
}
steps.push(drop_check_constraint_step(db_type, table_name, &check.name));
steps.push(add_check_constraint_step(db_type, table_name, check));
}
None => {
let semantic_match = actual.iter().any(|candidate| {
normalize_check_expression(&extract_check_expr(&candidate.definition))
== expected_norm
});
if semantic_match {
continue;
}
steps.push(add_check_constraint_step(db_type, table_name, check));
}
}
}
steps
}
#[cfg_attr(not(feature = "postgresql"), allow(dead_code))]
fn extract_check_expr(definition: &str) -> String {
let trimmed = definition.trim();
let body = trimmed
.strip_prefix("CHECK")
.map_or(trimmed, |rest| rest.trim());
body.to_string()
}
#[cfg_attr(not(feature = "postgresql"), allow(dead_code))]
fn drop_check_constraint_step(db_type: DbType, table_name: &str, constraint: &str) -> MigrationStep {
MigrationStep::Sql {
sql: format!(
"ALTER TABLE {} DROP CONSTRAINT {}",
crate::model::quote_qualified_identifier(db_type, table_name),
crate::model::quote_identifier(db_type, constraint)
),
}
}
#[cfg_attr(not(feature = "postgresql"), allow(dead_code))]
fn add_check_constraint_step(
db_type: DbType,
table_name: &str,
check: &ExpectedCheck,
) -> MigrationStep {
MigrationStep::AddConstraint {
table: table_name.to_string(),
definition: format!(
"CONSTRAINT {} CHECK ({})",
crate::model::quote_identifier(db_type, &check.name),
check.expr
),
}
}
#[cfg_attr(not(feature = "postgresql"), allow(dead_code))]
pub(crate) fn default_check_name(table_name: &str, column: &str) -> String {
format!(
"ck_{}_{}",
table_name.replace('.', "_"),
column
)
}
pub(crate) fn to_concurrent_create_index(sql: &str) -> String {
let upper = sql.to_ascii_uppercase();
if let Some(offset) = upper.find("CREATE UNIQUE INDEX") {
let mut out = sql.to_string();
out.insert_str(offset + "CREATE UNIQUE INDEX".len(), " CONCURRENTLY");
out
} else if let Some(offset) = upper.find("CREATE INDEX") {
let mut out = sql.to_string();
out.insert_str(offset + "CREATE INDEX".len(), " CONCURRENTLY");
out
} else {
sql.to_string()
}
}
fn is_index_creation_step(step: &MigrationStep) -> bool {
match step {
MigrationStep::CreateIndex { .. } => true,
MigrationStep::Sql { sql } => {
let upper = sql.trim_start().to_ascii_uppercase();
upper.starts_with("CREATE UNIQUE INDEX") || upper.starts_with("CREATE INDEX")
}
_ => false,
}
}
fn is_safe_extension_name(name: &str) -> bool {
!name.is_empty()
&& name
.chars()
.all(|ch| ch.is_ascii_alphanumeric() || ch == '_' || ch == '$')
&& !name
.chars()
.next()
.is_some_and(|first| first.is_ascii_digit())
}
impl Database {
pub async fn plan_table<T: WritableModel>(&self) -> crate::Result<TableDiagnosis> {
self.diagnose_family_merged::<T>(&ApplyOptions::default()).await
}
pub async fn apply_table<T: WritableModel>(
&self,
opts: &ApplyOptions,
) -> crate::Result<TableApplyOutcome> {
let db_type = self.db_type();
if !crate::abstract_layer::capabilities::Capabilities::of(db_type).schema_introspection {
return Err(crate::OrmerError::UnsupportedFeature {
backend: db_type,
feature: "apply_table",
});
}
self.ensure_extensions(&opts.extensions).await?;
let diagnosis = self.diagnose_family_merged::<T>(opts).await?;
let family = self.diagnose_family::<T>(opts).await?;
let table_name = T::table_name_for_db(db_type);
let mut executed: Vec<MigrationStep> = Vec::new();
let mut created_table = false;
let mut base_rebuild: Option<RebuildCause> = None;
match family.base {
TableDiagnosis::Ready => {}
TableDiagnosis::Migratable(plan) => {
created_table |= plan.steps().iter().any(|step| {
matches!(step, MigrationStep::CreateTable { table, .. } if table == table_name)
});
executed.extend(self.execute_plan_steps(&plan, opts).await?);
}
TableDiagnosis::NeedsRebuild(cause) => base_rebuild = Some(cause),
}
let mut child_rebuilt = false;
if base_rebuild.is_none() {
for (child, child_diagnosis) in family.children {
match child_diagnosis {
TableDiagnosis::Ready => {}
TableDiagnosis::Migratable(plan) => {
executed.extend(self.execute_plan_steps(&plan, opts).await?);
}
TableDiagnosis::NeedsRebuild(cause) => match opts.rebuild {
RebuildPolicy::Refuse => {
return Err(crate::OrmerError::unmigratable_schema(
cause.table,
cause.reason,
));
}
RebuildPolicy::Allow => {
#[cfg(feature = "postgresql")]
{
self.recreate_routed_child_table::<T>(&child).await?;
child_rebuilt = true;
}
#[cfg(not(feature = "postgresql"))]
{
let _ = &child;
let _ = &mut child_rebuilt;
}
}
},
}
}
}
if let Some(cause) = base_rebuild {
return self.rebuild_or_refuse::<T>(opts, cause).await;
}
match self.diagnose_family_merged::<T>(opts).await? {
TableDiagnosis::Ready => Ok(TableApplyOutcome {
diagnosis,
created_table: created_table || child_rebuilt,
executed,
}),
TableDiagnosis::Migratable(_) => {
let cause = RebuildCause::new(
table_name,
"incremental migration did not reconcile the schema; \
fixing it requires dropping and recreating the table",
);
self.rebuild_or_refuse::<T>(opts, cause).await
}
TableDiagnosis::NeedsRebuild(cause) => self.rebuild_or_refuse::<T>(opts, cause).await,
}
}
pub(crate) async fn ensure_extensions(&self, extensions: &[String]) -> crate::Result<()> {
if extensions.is_empty() {
return Ok(());
}
let db_type = self.db_type();
for name in extensions {
if !is_safe_extension_name(name) {
return Err(crate::ormer_error!(
"invalid extension name {name:?}: must be a plain identifier"
));
}
}
#[cfg(feature = "postgresql")]
{
if let Some(db) = self.as_postgresql() {
for name in extensions {
db.ensure_extension(name).await?;
}
return Ok(());
}
}
let _ = db_type;
Err(crate::OrmerError::UnsupportedFeature {
backend: db_type,
feature: "ApplyOptions.extensions (CREATE EXTENSION is PostgreSQL-only)",
})
}
pub async fn ping(&self) -> crate::Result<()> {
match self {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => db.ping().await,
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => db.ping().await,
#[cfg(feature = "mysql")]
Database::MySQL(db) => db.ping().await,
#[cfg(feature = "mssql")]
Database::MSSQL(db) => db.ping().await,
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => db.ping().await,
#[cfg(feature = "clickhouse")]
Database::ClickHouse(db) => db.ping().await,
#[cfg(feature = "influxdb")]
Database::InfluxDB(db) => db.ping().await,
}
}
pub(crate) async fn diagnose_single_table<T: WritableModel>(
&self,
table_name: Option<&str>,
opts: &ApplyOptions,
) -> crate::Result<TableDiagnosis> {
let db_type = self.db_type();
let base_name = T::table_name_for_db(db_type);
let target = table_name.unwrap_or(base_name);
let is_base = table_name.is_none();
let mut plan = MigrationPlan::new(target, db_type);
let exists = self.schema_columns(target).await?.is_some();
if !exists {
if !is_base {
return Ok(TableDiagnosis::Ready);
}
plan.push(MigrationStep::CreateTable {
table: target.to_string(),
definition: crate::generate_create_table_sql::<T>(db_type)?,
});
return Ok(TableDiagnosis::Migratable(plan));
}
let migration =
TableMigration::<T>::for_diagnosis(self, table_name, opts.extra_columns);
match migration.plan_for_table(target).await {
Ok(diff) => {
if diff.is_empty() {
Ok(TableDiagnosis::Ready)
} else {
plan.steps.extend(diff.steps);
plan.warnings.extend(diff.warnings);
Ok(TableDiagnosis::Migratable(plan))
}
}
Err(err) if err.is_unmigratable_schema() || is_schema_rebuild_error(&err) => {
Ok(TableDiagnosis::NeedsRebuild(RebuildCause::new(
target,
err.to_string(),
)))
}
Err(err @ crate::OrmerError::UnsupportedFeature { .. }) if db_type.is_questdb() => {
Ok(TableDiagnosis::NeedsRebuild(RebuildCause::new(
target,
err.to_string(),
)))
}
Err(err) => Err(err),
}
}
async fn diagnose_family<T: WritableModel>(
&self,
opts: &ApplyOptions,
) -> crate::Result<FamilyDiagnosis> {
let db_type = self.db_type();
if !crate::abstract_layer::capabilities::Capabilities::of(db_type).schema_introspection {
return Err(crate::OrmerError::UnsupportedFeature {
backend: db_type,
feature: "plan_table",
});
}
let base = self.diagnose_single_table::<T>(None, opts).await?;
#[cfg(feature = "postgresql")]
let mut children = Vec::new();
#[cfg(not(feature = "postgresql"))]
let children = Vec::new();
#[cfg(feature = "postgresql")]
{
if matches!(db_type, DbType::PostgreSQL) && T::hypertable_route_key().is_some() {
for child in self.existing_routed_child_tables::<T>().await? {
if self.schema_columns(&child).await?.is_none() {
continue;
}
let child_diagnosis = self
.diagnose_single_table::<T>(Some(&child), opts)
.await?;
children.push((child, child_diagnosis));
}
}
}
Ok(FamilyDiagnosis { base, children })
}
pub(crate) async fn diagnose_family_merged<T: WritableModel>(
&self,
opts: &ApplyOptions,
) -> crate::Result<TableDiagnosis> {
let family = self.diagnose_family::<T>(opts).await?;
let mut merged = family.base;
for (_, child) in family.children {
merged = merge_diagnoses(merged, child);
}
Ok(merged)
}
async fn rebuild_or_refuse<T: WritableModel>(
&self,
opts: &ApplyOptions,
cause: RebuildCause,
) -> crate::Result<TableApplyOutcome> {
let table_name = T::table_name_for_db(self.db_type());
match opts.rebuild {
RebuildPolicy::Refuse => Err(crate::OrmerError::unmigratable_schema(
cause.table,
cause.reason,
)),
RebuildPolicy::Allow => {
let triggers = self.backup_business_triggers(table_name).await?;
self.drop_table::<T>().execute().await?;
self.create_table::<T>().execute().await?;
self.restore_business_triggers(&triggers).await?;
match self.diagnose_family_merged::<T>(opts).await? {
TableDiagnosis::Ready => {
let db_type = self.db_type();
let mut executed = vec![
MigrationStep::Sql {
sql: format!(
"DROP TABLE {}",
crate::model::quote_qualified_identifier(db_type, table_name)
),
},
MigrationStep::CreateTable {
table: table_name.to_string(),
definition: crate::generate_create_table_sql::<T>(db_type)?,
},
];
for definition in &triggers {
executed.push(MigrationStep::Sql {
sql: definition.clone(),
});
}
Ok(TableApplyOutcome {
diagnosis: TableDiagnosis::NeedsRebuild(cause),
created_table: true,
executed,
})
}
other => Err(crate::ormer_error!(
"apply_table rebuild of {table_name} did not reconcile the schema: {other:?}"
)),
}
}
}
}
#[cfg_attr(not(feature = "postgresql"), allow(unused_variables))]
pub(crate) async fn execute_plan_steps(
&self,
plan: &MigrationPlan,
opts: &ApplyOptions,
) -> crate::Result<Vec<MigrationStep>> {
if plan.is_empty() {
return Ok(Vec::new());
}
let db_type = self.db_type();
#[cfg(any(feature = "sqlite", feature = "postgresql"))]
let concurrent = {
#[cfg(feature = "postgresql")]
{
opts.index_concurrently
&& matches!(db_type, DbType::PostgreSQL)
&& !db_type.is_questdb()
}
#[cfg(not(feature = "postgresql"))]
{
false
}
};
#[cfg(not(any(feature = "sqlite", feature = "postgresql")))]
let concurrent = false;
let all: Vec<MigrationStep> = plan.steps().to_vec();
if !db_type.is_transactional() {
execute_steps_nontransactional(self, db_type, plan.steps()).await?;
return Ok(all);
}
if !concurrent {
let mut transaction = self.begin().await?;
let result = execute_steps(&mut transaction, db_type, plan.steps()).await;
match result {
Ok(()) => transaction.commit().await?,
Err(error) => {
let _ = transaction.rollback().await;
return Err(error);
}
}
return Ok(all);
}
let mut executed = Vec::new();
let mut transaction: Option<Transaction<'_>> = Some(self.begin().await?);
for step in &all {
if is_index_creation_step(step) {
if let Some(open) = transaction.take() {
open.commit().await?;
}
self.create_index_concurrently(step, db_type).await?;
} else {
if transaction.is_none() {
transaction = Some(self.begin().await?);
}
let open = transaction
.as_mut()
.expect("transaction re-opened above");
if let Err(error) = execute_steps(open, db_type, std::slice::from_ref(step)).await
{
if let Some(open) = transaction.take() {
let _ = open.rollback().await;
}
return Err(error);
}
}
executed.push(step.clone());
}
if let Some(open) = transaction.take() {
open.commit().await?;
}
Ok(executed)
}
async fn create_index_concurrently(
&self,
step: &MigrationStep,
db_type: DbType,
) -> crate::Result<()> {
let sql = step.sql(db_type)?;
#[cfg(feature = "postgresql")]
{
if let Some(db) = self.as_postgresql() {
let (schema, bare) = split_qualified_table_name(table_name(step));
let schema = schema.unwrap_or("public");
if db.is_hypertable_table(schema, bare).await? {
self.execute_sql(sql.as_str()).await?;
return Ok(());
}
}
}
let concurrent_sql = to_concurrent_create_index(&sql);
if concurrent_sql == sql {
return Err(crate::ormer_error!(
"cannot render CREATE INDEX CONCURRENTLY for step: {sql}"
));
}
self.drop_invalid_indexes_for_table(db_type, table_name(step))
.await?;
match self.execute_sql(concurrent_sql.as_str()).await {
Ok(_) => Ok(()),
Err(first_error) => {
self.drop_invalid_indexes_for_table(db_type, table_name(step))
.await?;
match self.execute_sql(concurrent_sql.as_str()).await {
Ok(_) => Ok(()),
Err(retry_error) => Err(crate::ormer_error!(
"CREATE INDEX CONCURRENTLY failed after INVALID cleanup retry: {retry_error} (first attempt: {first_error})"
)),
}
}
}
}
pub(crate) async fn drop_invalid_indexes_for_table(
&self,
db_type: DbType,
table_name: &str,
) -> crate::Result<()> {
#[cfg(feature = "postgresql")]
{
if let Some(db) = self.as_postgresql() {
let (schema, bare) = split_qualified_table_name(table_name);
let schema = schema.unwrap_or("public");
let names = db.invalid_index_names(schema, bare).await?;
for name in names {
let qualified = format!(
"{}.{}",
crate::model::quote_identifier(db_type, schema),
crate::model::quote_identifier(db_type, &name)
);
self.execute_sql(format!("DROP INDEX IF EXISTS {qualified}").as_str())
.await?;
}
}
}
let _ = (db_type, table_name);
Ok(())
}
pub(crate) async fn backup_business_triggers(
&self,
qualified_table: &str,
) -> crate::Result<Vec<String>> {
#[cfg(feature = "postgresql")]
{
if let Some(db) = self.as_postgresql() {
let (schema, bare) = split_qualified_table_name(qualified_table);
let schema = schema.unwrap_or("public");
return db.business_trigger_definitions(schema, bare).await;
}
}
let _ = qualified_table;
Ok(Vec::new())
}
pub(crate) async fn restore_business_triggers(
&self,
definitions: &[String],
) -> crate::Result<()> {
for definition in definitions {
self.execute_sql(definition.as_str()).await?;
}
Ok(())
}
#[cfg(feature = "postgresql")]
pub(crate) fn as_postgresql(&self) -> Option<&crate::abstract_layer::postgresql_backend::Database> {
match self {
Database::PostgreSQL(db) if !self.db_type().is_questdb() => Some(db),
_ => None,
}
}
pub(crate) async fn actual_index_facts(
&self,
table_name: &str,
indexes: &[crate::db_first::DbFirstIndex],
) -> crate::Result<Vec<ActualIndexFacts>> {
#[cfg(feature = "postgresql")]
{
if let Some(db) = self.as_postgresql() {
let (schema, bare) = split_qualified_table_name(table_name);
let schema = schema.unwrap_or("public");
return db.index_facts(schema, bare).await;
}
}
let _ = table_name;
Ok(indexes
.iter()
.map(|index| {
ActualIndexFacts::from_plain(
index.name.clone(),
index
.columns
.iter()
.map(|column| (column.name.clone(), column.descending))
.collect(),
index.unique,
)
})
.collect())
}
pub(crate) async fn primary_key_constraint_name(
&self,
table_name: &str,
) -> crate::Result<Option<String>> {
#[cfg(feature = "postgresql")]
{
if let Some(db) = self.as_postgresql() {
let (schema, bare) = split_qualified_table_name(table_name);
let schema = schema.unwrap_or("public");
return db.primary_key_constraint_name(schema, bare).await;
}
}
let _ = table_name;
Ok(None)
}
#[cfg_attr(not(feature = "postgresql"), allow(dead_code))]
pub(crate) async fn check_constraint_facts(
&self,
schema: &str,
table: &str,
) -> crate::Result<Vec<ActualCheckFacts>> {
#[cfg(feature = "postgresql")]
{
if let Some(db) = self.as_postgresql() {
return db.check_constraint_facts(schema, table).await;
}
}
let _ = (schema, table);
Ok(Vec::new())
}
}
struct FamilyDiagnosis {
base: TableDiagnosis,
children: Vec<(String, TableDiagnosis)>,
}
fn merge_diagnoses(left: TableDiagnosis, right: TableDiagnosis) -> TableDiagnosis {
use TableDiagnosis::*;
match (left, right) {
(NeedsRebuild(cause), _) | (_, NeedsRebuild(cause)) => NeedsRebuild(cause),
(Migratable(mut plan), Migratable(other)) => {
plan.steps.extend(other.steps);
plan.warnings.extend(other.warnings);
Migratable(plan)
}
(Migratable(plan), Ready) => Migratable(plan),
(Ready, Migratable(plan)) => Migratable(plan),
(Ready, Ready) => Ready,
}
}
impl ExpectedIndexDef<'_> {
pub(crate) fn index_method(&self) -> Option<&'static str> {
self.columns.iter().find_map(|column| column.index_method)
}
pub(crate) fn index_expression(&self) -> Option<String> {
self.columns
.iter()
.find_map(|column| column.index_expression)
.map(str::to_string)
.or_else(|| {
self.columns
.iter()
.find_map(|column| column.index_columns)
.map(str::to_string)
})
}
pub(crate) fn index_predicate(&self) -> Option<&'static str> {
self.columns.iter().find_map(|column| column.index_where)
}
pub(crate) fn is_special(&self) -> bool {
self.index_method().is_some() || self.index_expression().is_some()
}
}
#[cfg(feature = "postgresql")]
fn postgres_truncated_identifier(name: &str) -> String {
if name.len() <= 63 {
return name.to_string();
}
let mut end = 63;
while !name.is_char_boundary(end) {
end -= 1;
}
name[..end].to_string()
}
pub(crate) fn plan_index_semantic_diff(
db_type: DbType,
table_name: &str,
expected: &[ExpectedIndexDef<'_>],
actual: &[ActualIndexFacts],
available_columns: &BTreeSet<&str>,
foreign_key_columns: &BTreeSet<&str>,
unique_name_reliable: bool,
) -> crate::Result<(Vec<MigrationStep>, Vec<String>)> {
let mut drops: Vec<MigrationStep> = Vec::new();
let mut creates: Vec<MigrationStep> = Vec::new();
let mut warnings: Vec<String> = Vec::new();
let mut consumed = vec![false; actual.len()];
for expected_def in expected {
let special = expected_def.is_special();
let method = expected_def.index_method();
let expression = expected_def.index_expression();
let predicate = expected_def.index_predicate();
let plain_columns: Vec<(String, bool)> = if special {
Vec::new()
} else {
expected_def
.columns
.iter()
.map(|column| {
(
column.name.to_string(),
column.index_order == Some("DESC"),
)
})
.collect()
};
let matched = actual.iter().enumerate().find(|(position, candidate)| {
!consumed[*position]
&& match_index_semantics(
expected_def.unique,
method,
&plain_columns,
expression.as_deref(),
predicate,
candidate,
)
});
match matched {
Some((position, actual_index)) => {
consumed[position] = true;
let ormer_name = expected_def
.name
.map(ToString::to_string)
.unwrap_or_else(|| default_index_name_fallback(table_name, expected_def));
let name_checks = unique_name_reliable || !expected_def.unique;
#[cfg(feature = "postgresql")]
let name_matches = actual_index.name == ormer_name
|| (db_type == DbType::PostgreSQL
&& actual_index.name == postgres_truncated_identifier(&ormer_name));
#[cfg(not(feature = "postgresql"))]
let name_matches = actual_index.name == ormer_name;
if name_checks && !name_matches {
warnings.push(format!(
"index {} on {table_name} matches the model semantically but uses a \
foreign name; renaming to {ormer_name} (drop + recreate)",
actual_index.name
));
drops.push(MigrationStep::DropIndex {
name: actual_index.name.clone(),
table: table_name.to_string(),
});
creates.push(create_index_step(
db_type,
&ormer_name,
table_name,
expected_def,
expression.as_deref(),
)?);
}
}
None => {
if !special
&& !expected_def
.columns
.iter()
.all(|column| available_columns.contains(column.name))
{
warnings.push(format!(
"skipping index on ({}) because some columns are missing from table {table_name}",
expected_def
.columns
.iter()
.map(|column| column.name)
.collect::<Vec<_>>()
.join(", ")
));
continue;
}
if special && actual_definition_unavailable(db_type) {
warnings.push(format!(
"special index on ({}) cannot be verified against this backend; \
its presence is not diffed automatically",
expected_def
.columns
.iter()
.map(|column| column.name)
.collect::<Vec<_>>()
.join(", ")
));
continue;
}
let name = expected_def
.name
.map(ToString::to_string)
.unwrap_or_else(|| default_index_name_fallback(table_name, expected_def));
creates.push(create_index_step(
db_type,
&name,
table_name,
expected_def,
expression.as_deref(),
)?);
}
}
}
let special_columns: BTreeSet<&str> = expected
.iter()
.filter(|expected_def| expected_def.is_special())
.flat_map(|expected_def| {
expected_def
.columns
.iter()
.map(|column| column.name)
.collect::<Vec<_>>()
})
.collect();
for (position, index) in actual.iter().enumerate() {
if consumed[position] || index.name.is_empty() {
continue;
}
if !index.valid {
drops.push(MigrationStep::DropIndex {
name: index.name.clone(),
table: table_name.to_string(),
});
continue;
}
let leading_column = index
.columns
.first()
.map(|(name, _)| name.clone())
.unwrap_or_default();
if !leading_column.is_empty() && foreign_key_columns.contains(leading_column.as_str()) {
warnings.push(format!(
"keeping index {} because it backs a foreign key on column {leading_column}",
index.name
));
continue;
}
if index.definition.is_none()
&& index
.columns
.iter()
.any(|(name, _)| special_columns.contains(name.as_str()))
{
warnings.push(format!(
"keeping index {} because it may implement a method/expression index declaration",
index.name
));
continue;
}
if index
.method
.as_deref()
.is_some_and(|method| !method.eq_ignore_ascii_case("btree"))
{
warnings.push(format!(
"keeping non-btree index {} because it is not declared in the model",
index.name
));
continue;
}
warnings.push(format!(
"dropping index {} because it is not declared in the model",
index.name
));
drops.push(MigrationStep::DropIndex {
name: index.name.clone(),
table: table_name.to_string(),
});
}
let mut steps = drops;
steps.extend(creates);
Ok((steps, warnings))
}
fn default_index_name_fallback(table_name: &str, expected: &ExpectedIndexDef<'_>) -> String {
crate::migration::default_index_name(table_name, expected)
}
fn actual_definition_unavailable(db_type: DbType) -> bool {
let has_definition_introspection = {
#[cfg(feature = "postgresql")]
{
matches!(db_type, DbType::PostgreSQL)
}
#[cfg(not(feature = "postgresql"))]
{
false
}
};
!has_definition_introspection && !crate::model::index_supports_if_not_exists(db_type)
}
fn create_index_step(
db_type: DbType,
name: &str,
table_name: &str,
expected: &ExpectedIndexDef<'_>,
expression: Option<&str>,
) -> crate::Result<MigrationStep> {
if expression.is_some() {
let columns_sql = expected
.columns
.iter()
.map(|column| {
let mut value = crate::model::quote_identifier(db_type, column.name);
if let Some(order) = column.index_order {
value.push(' ');
value.push_str(order);
}
value
})
.collect::<Vec<_>>();
let columns_body = match expression {
Some(expression_text) => expression_text.to_string(),
None => columns_sql.join(", "),
};
let unique_sql = if expected.unique { "UNIQUE " } else { "" };
let predicate = expected
.index_predicate()
.map(|where_clause| format!(" WHERE {where_clause}"))
.unwrap_or_default();
let base = crate::model::render_create_index(db_type, name, table_name, &columns_body, expected.unique);
let _ = (unique_sql, columns_sql);
return Ok(MigrationStep::Sql {
sql: format!("{base}{predicate}"),
});
}
crate::migration::index_migration_step(db_type, name.to_string(), table_name, &expected.columns, expected.unique)
}
#[cfg_attr(not(feature = "postgresql"), allow(unused_variables))]
pub(crate) fn plan_primary_key_change(
db_type: DbType,
table_name: &str,
actual_pk: &BTreeSet<String>,
expected_pk: &[&str],
expected_pk_constraint: Option<String>,
is_hypertable: bool,
partition_columns: &[&str],
) -> Result<Option<MigrationStep>, String> {
let expected_set: BTreeSet<&str> = expected_pk.iter().copied().collect();
if actual_pk.len() == expected_set.len()
&& actual_pk.iter().all(|name| expected_set.contains(name.as_str()))
{
return Ok(None);
}
#[cfg(feature = "postgresql")]
if matches!(db_type, DbType::PostgreSQL) {
if is_hypertable
&& partition_columns
.iter()
.any(|column| !expected_set.contains(column))
{
return Err(format!(
"hypertable primary key must contain partition column(s) [{}]; \
rebuilding the table is the only safe path",
partition_columns.join(", ")
));
}
return Ok(Some(MigrationStep::ChangePrimaryKey {
table: table_name.to_string(),
columns: expected_pk.iter().map(|column| column.to_string()).collect(),
drop_constraint: expected_pk_constraint,
}));
}
let _ = table_name;
Err("primary key change cannot be migrated in place on this backend".to_string())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn norm_sql_fragment_collapses_whitespace_and_quotes() {
assert_eq!(norm_sql_fragment("\"a\" , \"b\""), "a , b");
assert_eq!(norm_sql_fragment("( a )"), "a");
assert_eq!(norm_sql_fragment("(a, b)"), "(a, b)");
}
#[test]
fn parse_pg_index_definition_extracts_parts() {
let parsed = parse_pg_index_definition(
"CREATE INDEX idx ON public.t USING btree (a, b DESC) WHERE (deleted = false)",
)
.unwrap();
assert_eq!(parsed.method, "btree");
assert_eq!(parsed.columns, "a, b DESC");
assert_eq!(parsed.predicate.as_deref(), Some("(deleted = false)"));
let parsed = parse_pg_index_definition(
"CREATE UNIQUE INDEX uq ON public.t USING gin (to_tsvector('english'::regconfig, title))",
)
.unwrap();
assert_eq!(parsed.method, "gin");
assert_eq!(
parsed.columns,
"to_tsvector('english'::regconfig, title)"
);
assert_eq!(parsed.predicate, None);
}
#[test]
fn normalize_check_expression_strips_casts_and_identifier_parens() {
assert_eq!(
normalize_check_expression("CHECK ((cardinality((roles)::text[])) > 0)"),
"cardinality(roles) > 0"
);
assert_eq!(
normalize_check_expression("CHECK ((cardinality(roles)) > 0)"),
"cardinality(roles) > 0"
);
assert_eq!(
normalize_check_expression("CHECK ((status)::text = 'a::b')"),
"status = 'a::b'"
);
}
#[test]
fn check_diff_renders_on_sqlite() {
let expected = vec![ExpectedCheck {
name: "ck_t_roles".to_string(),
expr: "cardinality(roles) > 0".to_string(),
}];
let steps = plan_check_diff("t", &expected, &[], DbType::Sqlite);
assert_eq!(steps.len(), 1);
match &steps[0] {
MigrationStep::AddConstraint { definition, .. } => {
assert!(definition.contains("CHECK (cardinality(roles) > 0)"), "{definition}");
}
other => panic!("unexpected step: {other:?}"),
}
let actual = vec![ActualCheckFacts {
name: "ck_t_roles".to_string(),
definition: "CHECK ((cardinality((roles)::text[])) > 0)".to_string(),
}];
assert!(plan_check_diff("t", &expected, &actual, DbType::Sqlite).is_empty());
}
#[cfg(feature = "postgresql")]
#[test]
fn check_diff_skips_semantically_equal_constraints() {
let expected = vec![ExpectedCheck {
name: "ck_t_roles".to_string(),
expr: "cardinality(roles) > 0".to_string(),
}];
let actual = vec![ActualCheckFacts {
name: "ck_t_roles".to_string(),
definition: "CHECK ((cardinality((roles)::text[])) > 0)".to_string(),
}];
assert!(plan_check_diff("t", &expected, &actual, DbType::PostgreSQL).is_empty());
let renamed = vec![ActualCheckFacts {
name: "auth_event_roles_check".to_string(),
definition: "CHECK ((cardinality((roles)::text[])) > 0)".to_string(),
}];
assert!(plan_check_diff("t", &expected, &renamed, DbType::PostgreSQL).is_empty());
let drifted = vec![ActualCheckFacts {
name: "ck_t_roles".to_string(),
definition: "CHECK ((cardinality(roles)) >= 0)".to_string(),
}];
let steps = plan_check_diff("t", &expected, &drifted, DbType::PostgreSQL);
assert_eq!(steps.len(), 2);
assert!(matches!(steps[0], MigrationStep::Sql { .. }));
assert!(matches!(steps[1], MigrationStep::AddConstraint { .. }));
let steps = plan_check_diff("t", &expected, &[], DbType::PostgreSQL);
assert_eq!(steps.len(), 1);
assert!(matches!(steps[0], MigrationStep::AddConstraint { .. }));
}
#[test]
fn index_semantics_matches_by_columns_ignoring_name() {
let actual = ActualIndexFacts::from_plain(
"idx_auth_auth_users_project".to_string(),
vec![("project_id".to_string(), false)],
false,
);
assert!(match_index_semantics(
false,
None,
&[("project_id".to_string(), false)],
None,
None,
&actual
));
assert!(!match_index_semantics(
false,
None,
&[("update_time".to_string(), false), ("project_id".to_string(), false)],
None,
None,
&actual
));
}
#[test]
fn index_semantics_matches_expression_via_definition() {
let actual = ActualIndexFacts {
name: "idx_search".to_string(),
columns: vec![],
unique: false,
valid: true,
method: Some("gin".to_string()),
definition: Some(
"CREATE INDEX idx_search ON public.t USING gin (to_tsvector('english'::regconfig, title))"
.to_string(),
),
predicate: None,
};
assert!(match_index_semantics(
false,
Some("gin"),
&[],
Some("to_tsvector('english'::regconfig, title)"),
None,
&actual
));
assert!(!match_index_semantics(
false,
Some("gin"),
&[],
Some("to_tsvector('simple'::regconfig, title)"),
None,
&actual
));
}
#[test]
fn index_semantics_rejects_invalid_residue() {
let mut invalid = ActualIndexFacts::from_plain(
"idx_broken".to_string(),
vec![("a".to_string(), false)],
false,
);
invalid.valid = false;
assert!(!match_index_semantics(
false,
None,
&[("a".to_string(), false)],
None,
None,
&invalid
));
}
#[cfg(feature = "postgresql")]
#[test]
fn postgres_truncated_identifier_cuts_at_namedatalen() {
assert_eq!(
postgres_truncated_identifier("idx_t_a"),
"idx_t_a"
);
let long = "idx_collect_collect_report_task_event_exceptions_report_start_time";
assert_eq!(long.len(), 66);
assert_eq!(
postgres_truncated_identifier(long),
"idx_collect_collect_report_task_event_exceptions_report_start_t"
);
}
#[test]
fn concurrent_rewrite_covers_plain_and_unique_create_index() {
assert_eq!(
to_concurrent_create_index("CREATE INDEX idx_t_a ON t (a)"),
"CREATE INDEX CONCURRENTLY idx_t_a ON t (a)"
);
assert_eq!(
to_concurrent_create_index("CREATE UNIQUE INDEX uq_t_a ON t (a)"),
"CREATE UNIQUE INDEX CONCURRENTLY uq_t_a ON t (a)"
);
assert_eq!(
to_concurrent_create_index("ALTER TABLE t ADD COLUMN a int"),
"ALTER TABLE t ADD COLUMN a int"
);
}
#[cfg(feature = "postgresql")]
#[test]
fn primary_key_change_plan_on_postgresql() {
let mut actual = BTreeSet::new();
actual.insert("update_time".to_string());
actual.insert("project_id".to_string());
let expected = vec!["agvid", "update_time", "project_id"];
let step = plan_primary_key_change(
DbType::PostgreSQL,
"collect.data",
&actual,
&expected,
Some("data_pkey".to_string()),
false,
&[],
)
.unwrap()
.unwrap();
match step {
MigrationStep::ChangePrimaryKey {
columns,
drop_constraint,
..
} => {
assert_eq!(columns, vec!["agvid", "update_time", "project_id"]);
assert_eq!(drop_constraint.as_deref(), Some("data_pkey"));
}
other => panic!("unexpected step: {other:?}"),
}
let same = expected.iter().map(|c| c.to_string()).collect();
assert!(plan_primary_key_change(
DbType::PostgreSQL,
"collect.data",
&same,
&expected,
None,
false,
&[]
)
.unwrap()
.is_none());
assert!(plan_primary_key_change(
DbType::Sqlite,
"t",
&actual,
&expected,
None,
false,
&[]
)
.is_err());
}
#[cfg(feature = "postgresql")]
#[test]
fn primary_key_change_requires_partition_columns_for_hypertable() {
let mut actual = BTreeSet::new();
actual.insert("update_time".to_string());
let expected = vec!["project_id"];
assert!(plan_primary_key_change(
DbType::PostgreSQL,
"t",
&actual,
&expected,
None,
true,
&["update_time"]
)
.is_err());
let expected = vec!["project_id", "update_time"];
assert!(plan_primary_key_change(
DbType::PostgreSQL,
"t",
&actual,
&expected,
None,
true,
&["update_time"]
)
.unwrap()
.is_some());
}
}