safe_migrate/rules/
destructive.rs

1use crate::analysis::mutations::{AlterTableActionMutation, Mutation};
2use crate::analysis::state::{AnalysisState, CascadeResult, MutationResult};
3use crate::engine::config::Config;
4use crate::report::violations::{ObjectKind, OperationKind, Violation, ViolationTier};
5use crate::rules::Rule;
6
7pub const IRREVERSIBLE_MIGRATION_RULE_ID: &str = "irreversible-migration";
8
9pub struct CascadingDropRule;
10
11impl Rule for CascadingDropRule {
12    fn id(&self) -> &'static str {
13        "destructive-cascade"
14    }
15    fn default_tier(&self) -> ViolationTier {
16        ViolationTier::Tier1
17    }
18    fn recipe(&self) -> &'static str {
19        "Avoid CASCADE on DROP TABLE in production. Handle dependencies explicitly."
20    }
21
22    fn evaluate(
23        &self,
24        mutation: &Mutation,
25        result: &MutationResult,
26        pre_state: &crate::analysis::state::PreState,
27        state: &AnalysisState,
28        _config: &Config,
29        cascade_closure: Option<&CascadeResult>,
30    ) -> Vec<Violation> {
31        let mut violations = Vec::new();
32
33        if !matches!(result, MutationResult::Conflict { .. })
34            && let Mutation::DropTable(drop) = mutation
35            && drop.cascade
36            && let Some(closure) = cascade_closure
37        {
38            let mut affects_baseline = false;
39            let mut has_fk_pulled = false;
40
41            let roots = drop.ids.iter().collect::<std::collections::HashSet<_>>();
42            for rel_id in &closure.dropped_relations {
43                if !roots.contains(rel_id) && state.baseline_relations.contains(rel_id) {
44                    affects_baseline = true;
45                    if state.baseline_fk_dependencies.contains(rel_id) {
46                        has_fk_pulled = true;
47                    }
48                }
49            }
50
51            if !affects_baseline {
52                for (from_table, cname) in &closure.dropped_constraints {
53                    if pre_state
54                        .baseline_foreign_keys
55                        .contains(&(from_table.clone(), cname.clone()))
56                    {
57                        affects_baseline = true;
58                        break;
59                    }
60                }
61            }
62
63            for (from_table, _cname) in &closure.dropped_constraints {
64                if state.baseline_fk_dependencies.contains(from_table) {
65                    has_fk_pulled = true;
66                }
67            }
68
69            if affects_baseline {
70                let mut reason = format!(
71                    "DROP TABLE {} CASCADE silently destroys pre-existing database dependencies",
72                    drop.ids
73                        .iter()
74                        .map(ToString::to_string)
75                        .collect::<Vec<_>>()
76                        .join(", ")
77                );
78                if has_fk_pulled {
79                    reason.push_str(
80                        " (includes FK-pulled tables from other schemas — cross-team impact)",
81                    );
82                }
83                violations.push(Violation {
84                    source_range: None,
85                    rule_id: self.id(),
86                    operation_kind: OperationKind::DropTable,
87                    object_kind: ObjectKind::Table,
88                    object_name: drop
89                        .ids
90                        .iter()
91                        .map(ToString::to_string)
92                        .collect::<Vec<_>>()
93                        .join(", "),
94                    tier: self.default_tier(),
95                    reason,
96                    recipe: self.recipe(),
97                    dedup_key: None,
98                    sql: None,
99                    fk_dependency_related: has_fk_pulled,
100                });
101            }
102        }
103        violations
104    }
105}
106
107pub struct SizeAwareAddColumnRule;
108
109impl Rule for SizeAwareAddColumnRule {
110    fn id(&self) -> &'static str {
111        "size-aware-add-column"
112    }
113    fn default_tier(&self) -> ViolationTier {
114        ViolationTier::Tier1
115    }
116    fn recipe(&self) -> &'static str {
117        "Adding a column with a default requires a table rewrite. For PG11+, constant defaults are safe. For volatiles or <PG11, use a multi-step backfill."
118    }
119
120    fn evaluate(
121        &self,
122        mutation: &Mutation,
123        result: &MutationResult,
124        pre_state: &crate::analysis::state::PreState,
125        state: &AnalysisState,
126        config: &Config,
127        _cascade_closure: Option<&CascadeResult>,
128    ) -> Vec<Violation> {
129        if *result == MutationResult::Skipped {
130            return vec![];
131        }
132
133        let mut violations = Vec::new();
134        let pg_version = state.pg_version_num.unwrap_or(config.assume_pg_version);
135
136        if let Mutation::AlterTable(alter) = mutation
137            && let AlterTableActionMutation::AddColumn {
138                default: Some(def), ..
139            } = &alter.action
140        {
141            let is_volatile = def.is_volatile();
142            let requires_rewrite = is_volatile || pg_version < 110000;
143
144            if requires_rewrite {
145                let (has_wide_columns, is_stale, rows) = match pre_state.relations.get(&alter.id) {
146                    Some(rel) => {
147                        let wide = rel.columns.iter().any(|c| {
148                            c.avg_width.unwrap_or(0) >= config.toast_width_threshold_bytes
149                        });
150                        // Only cache-backed relations have meaningful statistics age.
151                        let stale = rel.is_stale() && state.baseline_relations.contains(&alter.id);
152                        (
153                            wide,
154                            stale,
155                            rel.estimated_rows.unwrap_or(config.default_rows),
156                        )
157                    }
158                    None => {
159                        // Missing relation metadata uses the conservative fallback.
160                        (false, true, config.default_rows)
161                    }
162                };
163
164                if is_stale {
165                    let key = format!("{}_stale_{}", self.id(), alter.id);
166                    violations.push(Violation { source_range: None,
167                        rule_id: self.id(),
168                        operation_kind: OperationKind::AddColumn,
169                        object_kind: ObjectKind::Table,
170                        object_name: alter.id.to_string(),
171                        tier: ViolationTier::Tier2,
172                        reason: format!(
173                            "Table {} statistics are stale. Lock evaluations may be inaccurate.",
174                            alter.id
175                        ),
176                        recipe: "Run ANALYZE to ensure accurate TOAST width and row estimates before structural changes.",
177                        dedup_key: Some(key),
178                                    sql: None,
179                                    fk_dependency_related: false,
180                    });
181                }
182
183                let tier1_threshold = config.rule_tier1_threshold(self.id());
184                let mut tier = if rows >= tier1_threshold {
185                    ViolationTier::Tier1
186                } else {
187                    ViolationTier::Tier2
188                };
189
190                if has_wide_columns && tier == ViolationTier::Tier2 {
191                    tier = ViolationTier::Tier1;
192                }
193
194                let mut reason = if is_volatile {
195                    format!(
196                        "Adding column with volatile DEFAULT to {} triggers a table rewrite",
197                        alter.id
198                    )
199                } else {
200                    format!(
201                        "Adding column with DEFAULT to {} triggers a table rewrite on Postgres < 11",
202                        alter.id
203                    )
204                };
205
206                if has_wide_columns && tier == ViolationTier::Tier1 {
207                    reason.push_str(" (Escalated due to wide TOAST columns)");
208                }
209                if is_stale {
210                    reason.push_str(" [WARNING: Based on unknown offline statistics]");
211                }
212
213                violations.push(Violation {
214                    source_range: None,
215                    rule_id: self.id(),
216                    operation_kind: OperationKind::AddColumn,
217                    object_kind: ObjectKind::Table,
218                    object_name: alter.id.to_string(),
219                    tier,
220                    reason,
221                    recipe: self.recipe(),
222                    dedup_key: None,
223                    sql: None,
224                    fk_dependency_related: false,
225                });
226            }
227        }
228        violations
229    }
230}
231
232pub struct DropDatabaseRule;
233
234impl Rule for DropDatabaseRule {
235    fn id(&self) -> &'static str {
236        "drop-database"
237    }
238    fn default_tier(&self) -> ViolationTier {
239        ViolationTier::Tier1
240    }
241    fn recipe(&self) -> &'static str {
242        "DROP DATABASE is an irreversible, high-blast-radius operation that destroys the entire database context."
243    }
244
245    fn evaluate(
246        &self,
247        mutation: &Mutation,
248        _result: &MutationResult,
249        _pre_state: &crate::analysis::state::PreState,
250        _state: &AnalysisState,
251        _config: &Config,
252        _cascade: Option<&CascadeResult>,
253    ) -> Vec<Violation> {
254        if let Mutation::DropDatabase(d) = mutation {
255            return vec![Violation {
256                source_range: None,
257                rule_id: self.id(),
258                operation_kind: OperationKind::DropDatabase,
259                object_kind: ObjectKind::Database,
260                object_name: d.id.to_string(),
261                tier: self.default_tier(),
262                reason: "DROP DATABASE detected".to_string(),
263                recipe: self.recipe(),
264                dedup_key: None,
265                sql: None,
266                fk_dependency_related: false,
267            }];
268        }
269        vec![]
270    }
271}
272
273pub struct DropSchemaCascadeRule;
274
275impl Rule for DropSchemaCascadeRule {
276    fn id(&self) -> &'static str {
277        "drop-schema-cascade"
278    }
279    fn default_tier(&self) -> ViolationTier {
280        ViolationTier::Tier1
281    }
282    fn recipe(&self) -> &'static str {
283        "DROP SCHEMA ... CASCADE recursively destroys every object in the schema. Handle dependencies explicitly."
284    }
285
286    fn evaluate(
287        &self,
288        mutation: &Mutation,
289        _result: &MutationResult,
290        _pre_state: &crate::analysis::state::PreState,
291        _state: &AnalysisState,
292        _config: &Config,
293        _cascade: Option<&CascadeResult>,
294    ) -> Vec<Violation> {
295        let mut violations = Vec::new();
296
297        if let Mutation::DropSchema(drop) = mutation
298            && drop.cascade
299        {
300            violations.push(Violation {
301                source_range: None,
302                rule_id: self.id(),
303                operation_kind: OperationKind::DropSchema,
304                object_kind: ObjectKind::Schema,
305                object_name: drop.names.join(", "),
306                tier: self.default_tier(),
307                reason: format!("DROP SCHEMA {} CASCADE detected", drop.names.join(", ")),
308                recipe: self.recipe(),
309                dedup_key: None,
310                sql: None,
311                fk_dependency_related: false,
312            });
313        }
314
315        violations
316    }
317}
318
319pub struct CreateTableAsSelectRule;
320
321impl Rule for CreateTableAsSelectRule {
322    fn id(&self) -> &'static str {
323        "create-table-as-select"
324    }
325    fn default_tier(&self) -> ViolationTier {
326        ViolationTier::Tier2
327    }
328    fn recipe(&self) -> &'static str {
329        "CREATE TABLE AS SELECT can be extremely slow and resource-intensive on large datasets. Consider creating the table first and using INSERT INTO ... SELECT in batches."
330    }
331
332    fn evaluate(
333        &self,
334        mutation: &Mutation,
335        result: &MutationResult,
336        _pre_state: &crate::analysis::state::PreState,
337        _state: &AnalysisState,
338        _config: &Config,
339        _cascade_closure: Option<&CascadeResult>,
340    ) -> Vec<Violation> {
341        if *result == MutationResult::Skipped {
342            return vec![];
343        }
344        if let Mutation::CreateTable(c) = mutation
345            && c.as_select
346        {
347            return vec![Violation {
348                source_range: None,
349                rule_id: self.id(),
350                operation_kind: OperationKind::CreateTable,
351                object_kind: ObjectKind::Table,
352                object_name: c.id.to_string(),
353                tier: self.default_tier(),
354                reason: format!("CREATE TABLE AS SELECT detected for {}", c.id),
355                recipe: self.recipe(),
356                dedup_key: None,
357                sql: None,
358                fk_dependency_related: false,
359            }];
360        }
361        vec![]
362    }
363}
364
365pub enum Reversibility {
366    Reversible,
367    ConditionallyReversible,
368    Irreversible,
369}
370
371pub fn classify(mutation: &Mutation) -> Reversibility {
372    match mutation {
373        Mutation::Rename(_) => Reversibility::Reversible,
374        Mutation::CreateIndex(_) | Mutation::CreateTable(_) => Reversibility::Reversible,
375        Mutation::AlterTable(a) => match &a.action {
376            AlterTableActionMutation::AddColumn { .. } => Reversibility::Reversible,
377            AlterTableActionMutation::DropColumn { .. } => Reversibility::Irreversible,
378            AlterTableActionMutation::SetType { .. } => Reversibility::ConditionallyReversible,
379            _ => Reversibility::Reversible,
380        },
381        Mutation::DropTable(_) | Mutation::DropDatabase(_) => Reversibility::Irreversible,
382        _ => Reversibility::ConditionallyReversible,
383    }
384}
385
386pub struct ReversibilityRule;
387
388impl Rule for ReversibilityRule {
389    fn id(&self) -> &'static str {
390        IRREVERSIBLE_MIGRATION_RULE_ID
391    }
392    fn default_tier(&self) -> ViolationTier {
393        ViolationTier::Tier1
394    }
395    fn recipe(&self) -> &'static str {
396        "This operation is irreversible. Ensure backups are available."
397    }
398
399    fn evaluate(
400        &self,
401        mutation: &Mutation,
402        result: &MutationResult,
403        pre_state: &crate::analysis::state::PreState,
404        state: &AnalysisState,
405        config: &Config,
406        _cascade_closure: Option<&CascadeResult>,
407    ) -> Vec<Violation> {
408        let skipped_known_drop_column = *result == MutationResult::Skipped
409            && matches!(
410                mutation,
411                Mutation::AlterTable(crate::analysis::mutations::AlterTable {
412                    id: _,
413                    action: AlterTableActionMutation::DropColumn { .. },
414                })
415            )
416            && matches!(
417                mutation,
418                Mutation::AlterTable(alter)
419                    if pre_state.relations.get(&alter.id).is_some_and(|relation| {
420                        matches!(
421                            &alter.action,
422                            AlterTableActionMutation::DropColumn { name, .. }
423                                if relation.has_column(name)
424                        )
425                    })
426            );
427        if *result != MutationResult::Applied && !skipped_known_drop_column {
428            return vec![];
429        }
430
431        let mut violations = Vec::new();
432
433        if let Mutation::AlterTable(a) = mutation
434            && let AlterTableActionMutation::SetType { column, ty, .. } = &a.action
435            && let Some(rel) = pre_state.relations.get(&a.id)
436            && let Some(old_ty) = rel.get_column(column).and_then(|c| c.data_type.as_ref())
437        {
438            // Only flag as "conditionally reversible" if there's actual data loss risk
439            // (e.g., narrowing, not widening). Table rewrites are handled by TypeChangeRewriteRule.
440            if is_type_change_lossy(old_ty, ty) {
441                let rows = rel.estimated_rows.unwrap_or(config.default_rows);
442                let tier = if rows >= config.rule_tier1_threshold(self.id()) {
443                    ViolationTier::Tier1
444                } else {
445                    ViolationTier::Tier2
446                };
447                violations.push(Violation {
448                    source_range: None,
449                    rule_id: self.id(),
450                    operation_kind: OperationKind::AlterColumnType,
451                    object_kind: ObjectKind::Table,
452                    object_name: a.id.to_string(),
453                    tier,
454                    reason: "Conditionally reversible type change detected".to_string(),
455                    recipe: "This type change may be lossy. Verify data compatibility.",
456                    dedup_key: None,
457                    sql: None,
458                    fk_dependency_related: false,
459                });
460            }
461        }
462
463        if let Reversibility::Irreversible = classify(mutation) {
464            // DropDatabaseRule owns this finding.
465            if matches!(mutation, Mutation::DropDatabase(_)) {
466                return violations;
467            }
468            let mut rows = if let Mutation::AlterTable(a) = mutation {
469                pre_state
470                    .relations
471                    .get(&a.id)
472                    .and_then(|r| r.estimated_rows)
473                    .unwrap_or(config.default_rows)
474            } else if let Mutation::DropTable(d) = mutation {
475                d.ids
476                    .iter()
477                    .filter_map(|id| pre_state.relations.get(id).and_then(|r| r.estimated_rows))
478                    .max()
479                    .unwrap_or(config.default_rows)
480            } else {
481                config.default_rows
482            };
483
484            if let Mutation::AlterTable(a) = mutation
485                && let AlterTableActionMutation::DropColumn { name, .. } = &a.action
486                && state.column_was_added_in_transaction(&a.id, name)
487            {
488                rows = 0;
489            }
490
491            let tier = if rows == 0 {
492                ViolationTier::Tier3
493            } else {
494                ViolationTier::Tier1
495            };
496
497            // Determine operation_kind and object_name from the mutation
498            let (operation_kind, object_kind, object_name) = match mutation {
499                Mutation::AlterTable(a) => match &a.action {
500                    AlterTableActionMutation::DropColumn { .. } => (
501                        OperationKind::DropColumn,
502                        ObjectKind::Table,
503                        a.id.to_string(),
504                    ),
505                    _ => (
506                        OperationKind::Irreversible,
507                        ObjectKind::Table,
508                        a.id.to_string(),
509                    ),
510                },
511                Mutation::DropTable(d) => (
512                    OperationKind::DropTable,
513                    ObjectKind::Table,
514                    d.ids
515                        .iter()
516                        .map(ToString::to_string)
517                        .collect::<Vec<_>>()
518                        .join(", "),
519                ),
520                Mutation::DropDatabase(d) => (
521                    OperationKind::DropDatabase,
522                    ObjectKind::Database,
523                    d.id.to_string(),
524                ),
525                _ => (
526                    OperationKind::Irreversible,
527                    ObjectKind::Table,
528                    "unknown".to_string(),
529                ),
530            };
531
532            violations.push(Violation {
533                source_range: None,
534                rule_id: self.id(),
535                operation_kind,
536                object_kind,
537                object_name,
538                tier,
539                reason: "Irreversible data-destructive operation detected".to_string(),
540                recipe: self.recipe(),
541                dedup_key: None,
542                sql: None,
543                fk_dependency_related: false,
544            });
545        }
546        violations
547    }
548}
549
550/// Checks if a type change represents actual data loss (narrowing), not just a rewrite.
551/// This is used by ReversibilityRule to distinguish between "safe widening" (e.g., INT->BIGINT)
552/// and genuinely lossy changes (e.g., BIGINT->INT, VARCHAR(255)->VARCHAR(50), TEXT->VARCHAR(n)).
553fn is_type_change_lossy(old_type: &str, new_type: &str) -> bool {
554    let old = old_type.to_lowercase().trim().to_string();
555    let new = new_type.to_lowercase().trim().to_string();
556
557    // Same type is trivially safe
558    if old == new {
559        return false;
560    }
561
562    // Extract base types (without parameters)
563    let old_base = old.split('(').next().unwrap_or(&old).trim();
564    let new_base = new.split('(').next().unwrap_or(&new).trim();
565
566    // VARCHAR narrowing check
567    if let (Some(old_lim), Some(new_lim)) =
568        (extract_varchar_limit(&old), extract_varchar_limit(&new))
569    {
570        // Both are varchar - check if narrowing
571        return new_lim < old_lim;
572    }
573
574    // Check for TEXT -> VARCHAR(n) narrowing (text is unbounded, varchar(n) is bounded)
575    // Also handles character varying (unbounded) -> varchar(n)
576    let new_varchar_limit = extract_varchar_limit(&new);
577    if new_varchar_limit.is_some()
578        && (old_base == "text" || old_base == "varchar" || old_base == "character varying")
579    {
580        return true;
581    }
582
583    // Varchar to something smaller/narrower - check if target type can hold all values
584    let old_varchar_limit = extract_varchar_limit(&old);
585    if old_varchar_limit.is_some() {
586        // varchar -> text is safe (widening)
587        if new == "text" || new == "varchar" || new == "character varying" {
588            return false;
589        }
590        // varchar -> other types might be lossy
591        return true;
592    }
593
594    // Widening integer types are safe (though they require a rewrite)
595    // int2 (smallint) -> int4 -> int8 (bigint) are all safe
596    if let (Some(old_sz), Some(new_sz)) = (
597        integer_type_size_bits(old_base),
598        integer_type_size_bits(new_base),
599    ) {
600        // Narrowing is unsafe
601        return new_sz < old_sz;
602    }
603
604    // For other types, assume safe unless we have specific knowledge
605    false
606}
607
608/// Extracts the character limit from a varchar type string.
609/// Returns the limit in bytes (for comparison purposes).
610fn extract_varchar_limit(ty: &str) -> Option<i32> {
611    if ty.starts_with("varchar(") || ty.starts_with("character varying(") {
612        let paren_start = ty.find('(')?;
613        let paren_end = ty[paren_start..].find(')')?;
614        let num_str = &ty[paren_start + 1..paren_start + paren_end];
615        let limit: i32 = num_str.parse().ok()?;
616        Some(limit)
617    } else if ty == "varchar" || ty == "character varying" {
618        // VARCHAR without limit is like TEXT - unbounded
619        None
620    } else {
621        None
622    }
623}
624
625/// Returns the size of integer types in bits.
626fn integer_type_size_bits(ty: &str) -> Option<i32> {
627    match ty {
628        "smallint" | "int2" => Some(16),
629        "integer" | "int4" | "int" => Some(32),
630        "bigint" | "int8" => Some(64),
631        _ => None,
632    }
633}
634
635pub struct GeneralCascadeRule;
636
637impl Rule for GeneralCascadeRule {
638    fn id(&self) -> &'static str {
639        "destructive-general-cascade"
640    }
641    fn default_tier(&self) -> ViolationTier {
642        ViolationTier::Tier1
643    }
644    fn recipe(&self) -> &'static str {
645        "Using CASCADE on DROP operations can silently delete dependent objects. Explicitly drop dependencies to avoid accidental data loss."
646    }
647
648    fn evaluate(
649        &self,
650        mutation: &Mutation,
651        _result: &MutationResult,
652        _pre_state: &crate::analysis::state::PreState,
653        _state: &AnalysisState,
654        _config: &Config,
655        _cascade: Option<&CascadeResult>,
656    ) -> Vec<Violation> {
657        let cascade_info: Option<(OperationKind, ObjectKind, String)> = match mutation {
658            Mutation::DropView(d) if d.cascade => Some((
659                OperationKind::DropView,
660                ObjectKind::View,
661                d.ids
662                    .iter()
663                    .map(|id| id.to_string())
664                    .collect::<Vec<_>>()
665                    .join(", "),
666            )),
667            Mutation::DropMaterializedView(d) if d.cascade => Some((
668                OperationKind::DropMaterializedView,
669                ObjectKind::MaterializedView,
670                d.ids
671                    .iter()
672                    .map(|id| id.to_string())
673                    .collect::<Vec<_>>()
674                    .join(", "),
675            )),
676            Mutation::DropSequence(d) if d.cascade => Some((
677                OperationKind::DropSequence,
678                ObjectKind::Sequence,
679                d.ids
680                    .iter()
681                    .map(|id| id.to_string())
682                    .collect::<Vec<_>>()
683                    .join(", "),
684            )),
685            Mutation::DropDomain(d) if d.cascade => Some((
686                OperationKind::DropDomain,
687                ObjectKind::Domain,
688                d.ids
689                    .iter()
690                    .map(|id| id.to_string())
691                    .collect::<Vec<_>>()
692                    .join(", "),
693            )),
694            Mutation::DropFunction(d) if d.cascade => Some((
695                OperationKind::DropFunction,
696                ObjectKind::Function,
697                "function".to_string(),
698            )),
699            Mutation::DropProcedure(d) if d.cascade => Some((
700                OperationKind::DropProcedure,
701                ObjectKind::Procedure,
702                "procedure".to_string(),
703            )),
704            Mutation::DropPublication(d) if d.cascade => Some((
705                OperationKind::DropPublication,
706                ObjectKind::Publication,
707                d.names.join(", "),
708            )),
709            _ => None,
710        };
711
712        if let Some((operation_kind, object_kind, object_name)) = cascade_info {
713            return vec![Violation {
714                source_range: None,
715                rule_id: self.id(),
716                operation_kind,
717                object_kind,
718                object_name,
719                tier: self.default_tier(),
720                reason: "Destructive CASCADE operation detected".to_string(),
721                recipe: self.recipe(),
722                dedup_key: None,
723                sql: None,
724                fk_dependency_related: false,
725            }];
726        }
727        vec![]
728    }
729}
730
731pub struct TypeChangeRewriteRule;
732
733impl TypeChangeRewriteRule {
734    fn is_type_change_safe(old_type: &str, new_type: &str, pg_version: u32) -> bool {
735        let old = old_type.to_lowercase();
736        let new = new_type.to_lowercase();
737        if old == new {
738            return true;
739        }
740
741        let old_base = old.split('(').next().unwrap_or(&old).trim();
742        let new_base = new.split('(').next().unwrap_or(&new).trim();
743
744        if (old_base == "varchar" || old_base == "character varying")
745            && (new_base == "varchar" || new_base == "character varying" || new_base == "text")
746        {
747            if new == "text" || new == "varchar" || new == "character varying" {
748                return true;
749            }
750            if let Some(old_mod) = extract_type_modifier_from_type_string(&old)
751                && let Some(new_mod) = extract_type_modifier_from_type_string(&new)
752                && old_mod <= new_mod
753            {
754                return true;
755            }
756        }
757
758        if pg_version >= 120000
759            && (old_base == "numeric" || old_base == "decimal")
760            && (new_base == "numeric" || new_base == "decimal")
761        {
762            // Changing to unconstrained numeric is always safe (widest form).
763            if !new.contains('(') {
764                return true;
765            }
766            // Changing to a constrained numeric(p, s) is safe if the new
767            // precision is >= the old precision and the new scale is >= the
768            // old scale (i.e. the new type can represent every value the old
769            // type could). If the old type is unconstrained we cannot prove
770            // widening, so we fall through to false.
771            if let (Some((old_p, old_s)), Some((new_p, new_s))) =
772                (parse_numeric_params(&old), parse_numeric_params(&new))
773                && new_p >= old_p
774                && new_s >= old_s
775            {
776                return true;
777            }
778        }
779
780        false
781    }
782
783    /// Detects whether a type change narrows a VARCHAR(n) column
784    /// using type_modifier values from the cache.
785    ///
786    /// atttypmod for VARCHAR(n) encodes the character limit:
787    ///   typmod = (limit + VARHDRSZ), where VARHDRSZ is 4
788    ///
789    /// A smaller typmod means a smaller character limit, which is lossy.
790    /// Returns true if the new modifier represents a smaller limit than the old.
791    pub fn is_lossy_varchar_narrowing(
792        old_modifier: Option<i32>,
793        new_modifier: Option<i32>,
794    ) -> bool {
795        match (old_modifier, new_modifier) {
796            // PostgreSQL uses -1 for an unbounded character limit.
797            (Some(-1), Some(new)) if new != -1 => true,
798            // If the new one is unbounded, it's never narrowing
799            (_, Some(-1)) => false,
800            // Bounded values narrow when the new character limit is smaller.
801            (Some(old), Some(new)) => new < old,
802            // A missing modifier cannot prove a bounded old limit.
803            (None, Some(new)) if new != -1 => true,
804            _ => false,
805        }
806    }
807}
808
809/// Parses precision and scale from a numeric/decimal type string.
810/// Returns `Some((precision, scale))` for `numeric(p, s)` or `numeric(p)` (scale=0).
811/// Returns `None` if the type has no parameters.
812fn parse_numeric_params(ty: &str) -> Option<(i32, i32)> {
813    let lower = ty.to_lowercase();
814    let paren_start = lower.find('(')?;
815    let paren_end = lower.find(')')?;
816    let inner = &lower[paren_start + 1..paren_end];
817    let mut parts = inner.splitn(2, ',');
818    let precision: i32 = parts.next()?.trim().parse().ok()?;
819    let scale: i32 = parts
820        .next()
821        .map(|s| s.trim().parse().unwrap_or(0))
822        .unwrap_or(0);
823    Some((precision, scale))
824}
825
826/// Extracts a synthetic type_modifier-like value from a type string.
827/// Used when the new type comes from the migration SQL (not from the cache).
828/// For varchar(N), derives the atttypmod from the character limit.
829pub fn extract_type_modifier_from_type_string(ty: &str) -> Option<i32> {
830    let lower = ty.to_lowercase().trim().to_string();
831    // Check for varchar(N) or character varying(N)
832    if lower.starts_with("varchar(") || lower.starts_with("character varying(") {
833        let paren_start = lower.find('(')?;
834        let paren_end = lower[paren_start..].find(')')?;
835        let num_str = &lower[paren_start + 1..paren_start + paren_end];
836        let limit: i32 = num_str.parse().ok()?;
837        // VARCHAR atttypmod is the character limit plus VARHDRSZ.
838        Some(limit + 4)
839    } else {
840        None
841    }
842}
843
844impl Rule for TypeChangeRewriteRule {
845    fn id(&self) -> &'static str {
846        "type-change-rewrite"
847    }
848    fn default_tier(&self) -> ViolationTier {
849        ViolationTier::Tier1
850    }
851    fn recipe(&self) -> &'static str {
852        "Changing this column type requires an ACCESS EXCLUSIVE table rewrite. Add a new column, backfill, and swap."
853    }
854
855    fn evaluate(
856        &self,
857        mutation: &Mutation,
858        result: &MutationResult,
859        pre_state: &crate::analysis::state::PreState,
860        state: &AnalysisState,
861        config: &Config,
862        _cascade_closure: Option<&CascadeResult>,
863    ) -> Vec<Violation> {
864        // A skipped type change against a present relation can mean that the
865        // baseline did not expose enough column/type evidence to mutate state.
866        // Keep the rewrite warning conservative in that case; silently
867        // dropping it would turn an uncertain ACCESS EXCLUSIVE operation into
868        // a false clean result. Other skipped mutations remain non-events.
869        if *result == MutationResult::Skipped
870            && !matches!(
871                mutation,
872                Mutation::AlterTable(crate::analysis::mutations::AlterTable {
873                    action: AlterTableActionMutation::SetType { .. },
874                    ..
875                })
876            )
877        {
878            return vec![];
879        }
880
881        let mut violations = Vec::new();
882
883        if let Mutation::AlterTable(alter) = mutation
884            && let AlterTableActionMutation::SetType {
885                column,
886                ty,
887                has_using: _,
888            } = &alter.action
889        {
890            let pg_version = state.pg_version_num.unwrap_or(config.assume_pg_version);
891
892            let (is_safe, rows, old_type_str, old_modifier) =
893                match pre_state.relations.get(&alter.id) {
894                    Some(rel) => {
895                        let col_info = rel.columns.iter().find(|c| c.name == *column);
896                        let old_ty = col_info.and_then(|col| col.data_type.as_ref());
897
898                        let safe = old_ty
899                            .map(|o| Self::is_type_change_safe(o, ty, pg_version))
900                            .unwrap_or(false);
901                        (
902                            safe,
903                            rel.estimated_rows.unwrap_or(config.default_rows),
904                            old_ty.cloned().unwrap_or_else(|| "unknown".to_string()),
905                            col_info.and_then(|col| col.type_modifier),
906                        )
907                    }
908                    None => (false, config.default_rows, "unknown".to_string(), None),
909                };
910
911            if !is_safe {
912                let tier1_threshold = config.rule_tier1_threshold(self.id());
913
914                let tier = if rows >= tier1_threshold {
915                    ViolationTier::Tier1
916                } else {
917                    ViolationTier::Tier2
918                };
919
920                let new_modifier = extract_type_modifier_from_type_string(ty);
921
922                if Self::is_lossy_varchar_narrowing(old_modifier, new_modifier) {
923                    violations.push(Violation { source_range: None,
924                        rule_id: self.id(),
925                        operation_kind: OperationKind::AlterColumnType,
926                        object_kind: ObjectKind::Table,
927                        object_name: format!("{}.{}", alter.id, column),
928                        tier,
929                        reason: format!(
930                            "Changing column {}.{} type from {} to {} narrows VARCHAR precision (lossy)",
931                            alter.id, column, old_type_str, ty
932                        ),
933                        recipe: "Narrowing VARCHAR(n) precision may cause data truncation. Consider adding a new column, backfilling, and then dropping the old one.",
934                        dedup_key: None,
935                                    sql: None,
936                                    fk_dependency_related: false,
937                    });
938                } else {
939                    violations.push(Violation {
940                        source_range: None,
941                        rule_id: self.id(),
942                        operation_kind: OperationKind::AlterColumnType,
943                        object_kind: ObjectKind::Table,
944                        object_name: format!("{}.{}", alter.id, column),
945                        tier,
946                        reason: format!(
947                            "Changing column {}.{} type from {} to {} causes a table rewrite",
948                            alter.id, column, old_type_str, ty
949                        ),
950                        recipe: self.recipe(),
951                        dedup_key: None,
952                        sql: None,
953                        fk_dependency_related: false,
954                    });
955                }
956            }
957        }
958        violations
959    }
960}
safe_migrate/rules/destructive.rs

safe_migrate/rules/
destructive.rs
