diff --git a/lib/migration_generator/migration_generator.ex b/lib/migration_generator/migration_generator.ex
index c90c7439..381853fa 100644
--- a/lib/migration_generator/migration_generator.ex
+++ b/lib/migration_generator/migration_generator.ex
@@ -7,7 +7,7 @@ defmodule AshPostgres.MigrationGenerator do
 
   require Logger
 
-  alias AshPostgres.MigrationGenerator.{Operation, Phase}
+  alias AshPostgres.MigrationGenerator.{Operation, OperationCycleError, OperationDeps, Phase}
 
   defstruct snapshot_path: nil,
             migration_path: nil,
@@ -544,7 +544,7 @@ defmodule AshPostgres.MigrationGenerator do
         Enum.map(rename_map, fn {{new_table, schema}, %{old_table: old_table, snapshot: snapshot}} ->
           %Operation.RenameTable{
             old_table: old_table,
-            new_table: new_table,
+            table: new_table,
             schema: schema,
             multitenancy: snapshot.multitenancy,
             repo: repo
@@ -980,7 +980,7 @@ defmodule AshPostgres.MigrationGenerator do
 
   defp organize_operations(operations) do
     operations
-    |> sort_operations()
+    |> toposort_operations()
     |> streamline()
     |> group_into_phases()
     |> clean_phases()
@@ -1628,12 +1628,8 @@ defmodule AshPostgres.MigrationGenerator do
          acc
        ) do
     rest
-    |> Enum.take_while(fn
-      %custom{} when custom in [Operation.AddCustomStatement, Operation.RemoveCustomStatement] ->
-        false
-
-      op ->
-        op.table == table && op.schema == schema
+    |> Enum.take_while(fn op ->
+      op.table == table && op.schema == schema
     end)
     |> Enum.with_index()
     |> Enum.find(fn
@@ -1783,645 +1779,123 @@ defmodule AshPostgres.MigrationGenerator do
     group_into_phases(operations, nil, [phase | acc])
   end
 
-  defp sort_operations(ops, acc \\ [])
-  defp sort_operations([], acc), do: acc
-
-  defp sort_operations([op | rest], []), do: sort_operations(rest, [op])
-
-  defp sort_operations([op | rest], acc) do
-    acc = Enum.reverse(acc)
-
-    after_index = Enum.find_index(acc, &after?(op, &1))
-
-    new_acc =
-      if after_index do
-        acc
-        |> List.insert_at(after_index, op)
-        |> Enum.reverse()
-      else
-        [op | Enum.reverse(acc)]
-      end
-
-    sort_operations(rest, new_acc)
-  end
-
-  defp after?(_, %Operation.AlterDeferrability{direction: :down}), do: true
-  defp after?(%Operation.AlterDeferrability{direction: :up}, _), do: true
-
-  defp after?(
-         %Operation.RemovePrimaryKey{},
-         %Operation.DropForeignKey{}
-       ),
-       do: true
-
-  defp after?(
-         %Operation.DropForeignKey{},
-         %Operation.RemovePrimaryKey{}
-       ),
-       do: false
-
-  defp after?(%Operation.RemovePrimaryKey{}, _), do: false
-  defp after?(_, %Operation.RemovePrimaryKey{}), do: true
-  defp after?(%Operation.RemovePrimaryKeyDown{}, _), do: true
-  defp after?(_, %Operation.RemovePrimaryKeyDown{}), do: false
-
-  defp after?(%Operation.AddPrimaryKeyDown{}, _), do: false
-  defp after?(_, %Operation.AddPrimaryKeyDown{}), do: true
-
-  defp after?(%Operation.AddPrimaryKey{}, _), do: true
-  defp after?(_, %Operation.AddPrimaryKey{}), do: false
-
-  defp after?(
-         %Operation.AddCustomStatement{},
-         _
-       ),
-       do: true
-
-  defp after?(
-         _,
-         %Operation.RemoveCustomStatement{}
-       ),
-       do: true
-
-  defp after?(
-         %Operation.AddAttribute{attribute: %{order: l}, table: table, schema: schema},
-         %Operation.AddAttribute{attribute: %{order: r}, table: table, schema: schema}
-       ),
-       do: l > r
-
-  defp after?(
-         %Operation.RenameUniqueIndex{
-           table: table,
-           schema: schema
-         },
-         %{table: table, schema: schema}
-       ) do
-    true
-  end
-
-  # CreateTable must appear before AddUniqueIndex for the same table (table must exist first).
-  defp after?(
-         %Operation.CreateTable{table: table, schema: schema},
-         %Operation.AddUniqueIndex{table: table, schema: schema}
-       ),
-       do: false
-
-  # Unique index must be created before any alter that adds FKs referencing it.
-  defp after?(
-         %Operation.AddUniqueIndex{table: table, schema: schema},
-         %Operation.AlterAttribute{table: table, schema: schema}
-       ),
-       do: false
-
-  # AddUniqueIndex must come after CreateTable (table must exist first).
-  defp after?(
-         %Operation.AddUniqueIndex{table: table, schema: schema},
-         %Operation.CreateTable{table: table, schema: schema}
-       ),
-       do: true
-
-  # Place AddUniqueIndex after a specific attribute (by source) for the same
-  # table so it appears before AlterAttributes (issue #236).
-  defp after?(
-         %Operation.AddUniqueIndex{
-           insert_after_attribute_source: source,
-           table: table,
-           schema: schema
-         },
-         %Operation.AddAttribute{
-           table: table,
-           schema: schema,
-           attribute: %{source: source}
-         }
-       )
-       when not is_nil(source),
-       do: true
-
-  defp after?(
-         %Operation.AddUniqueIndex{
-           insert_after_attribute_source: source,
-           table: table,
-           schema: schema
-         },
-         %Operation.AddAttribute{
-           table: table,
-           schema: schema
-         }
-       )
-       when not is_nil(source),
-       do: true
-
-  defp after?(
-         %Operation.AddUniqueIndex{
-           identity: %{keys: keys},
-           table: table,
-           schema: schema
-         },
-         %Operation.AlterAttribute{
-           table: table,
-           schema: schema,
-           new_attribute: %{
-             references: %{table: table, destination_attribute: destination_attribute}
-           }
-         }
-       ) do
-    destination_attribute not in List.wrap(keys)
-  end
-
-  defp after?(
-         %Operation.AddUniqueIndex{
-           table: table,
-           schema: schema
-         },
-         %{table: table, schema: schema}
-       ) do
-    true
-  end
-
-  defp after?(
-         %Operation.AddReferenceIndex{
-           table: table,
-           schema: schema
-         },
-         %{table: table, schema: schema}
-       ) do
-    true
-  end
-
-  defp after?(
-         %Operation.AddCheckConstraint{
-           constraint: %{attribute: attribute_or_attributes},
-           table: table,
-           multitenancy: multitenancy,
-           schema: schema
-         },
-         %Operation.AddAttribute{table: table, attribute: %{source: source}, schema: schema}
-       ) do
-    source in List.wrap(attribute_or_attributes) ||
-      (multitenancy.attribute && multitenancy.attribute in List.wrap(attribute_or_attributes))
-  end
-
-  defp after?(
-         %Operation.AddCustomIndex{
-           table: table,
-           schema: schema
-         },
-         %Operation.AddAttribute{table: table, schema: schema}
-       ) do
-    true
-  end
-
-  defp after?(
-         %Operation.AddCustomIndex{
-           table: table,
-           schema: schema
-         },
-         %Operation.RenameAttribute{
-           table: table,
-           schema: schema
-         }
-       ) do
-    true
-  end
-
-  defp after?(
-         %Operation.AddReferenceIndex{
-           table: table,
-           schema: schema
-         },
-         %Operation.AddAttribute{table: table, schema: schema}
-       ) do
-    true
-  end
-
-  defp after?(
-         %Operation.AddCustomIndex{
-           table: table,
-           schema: schema,
-           index: %{
-             concurrently: true
-           }
-         },
-         %Operation.AddCustomIndex{
-           table: table,
-           schema: schema,
-           index: %{
-             concurrently: false
-           }
-         }
-       ) do
-    true
-  end
-
-  defp after?(
-         %Operation.AddCheckConstraint{table: table, schema: schema, constraint: %{name: name}},
-         %Operation.RemoveCheckConstraint{
-           table: table,
-           schema: schema,
-           constraint: %{
-             name: name
-           }
-         }
-       ),
-       do: true
-
-  defp after?(
-         %Operation.RemoveCheckConstraint{
-           table: table,
-           schema: schema,
-           constraint: %{
-             name: name
-           }
-         },
-         %Operation.AddCheckConstraint{table: table, schema: schema, constraint: %{name: name}}
-       ),
-       do: false
-
-  defp after?(
-         %Operation.AddCheckConstraint{
-           constraint: %{attribute: attribute_or_attributes},
-           table: table,
-           schema: schema
-         },
-         %Operation.AlterAttribute{table: table, new_attribute: %{source: source}, schema: schema}
-       ) do
-    source in List.wrap(attribute_or_attributes)
-  end
-
-  defp after?(
-         %Operation.AddCheckConstraint{
-           constraint: %{attribute: attribute_or_attributes},
-           table: table,
-           schema: schema
-         },
-         %Operation.RenameAttribute{
-           table: table,
-           new_attribute: %{source: source},
-           schema: schema
-         }
-       ) do
-    source in List.wrap(attribute_or_attributes)
-  end
-
-  defp after?(
-         %Operation.RemoveUniqueIndex{table: table, schema: schema},
-         %Operation.AddUniqueIndex{table: table, schema: schema}
-       ) do
-    false
-  end
-
-  defp after?(
-         %Operation.RemoveCustomIndex{table: table, schema: schema},
-         %Operation.AddCustomIndex{table: table, schema: schema}
-       ) do
-    false
-  end
-
-  defp after?(
-         %Operation.RenameAttribute{
-           old_attribute: %{source: source},
-           table: table,
-           schema: schema
-         },
-         %Operation.RemoveUniqueIndex{
-           identity: %{keys: keys},
-           table: table,
-           schema: schema
-         }
-       ) do
-    source in List.wrap(keys)
-  end
-
-  defp after?(
-         %Operation.RemoveUniqueIndex{table: table, schema: schema},
-         %{table: table, schema: schema}
-       ) do
-    true
-  end
-
-  defp after?(
-         %Operation.RemoveCheckConstraint{
-           constraint: %{attribute: attributes},
-           table: table,
-           schema: schema
-         },
-         %Operation.RemoveAttribute{table: table, attribute: %{source: source}, schema: schema}
-       ) do
-    source in List.wrap(attributes)
-  end
-
-  defp after?(
-         %Operation.RemoveCheckConstraint{
-           constraint: %{attribute: attributes},
-           table: table,
-           schema: schema
-         },
-         %Operation.RenameAttribute{
-           table: table,
-           old_attribute: %{source: source},
-           schema: schema
-         }
-       ) do
-    source in List.wrap(attributes)
-  end
-
-  defp after?(%Operation.AlterAttribute{table: table, schema: schema}, %Operation.DropForeignKey{
-         table: table,
-         schema: schema,
-         direction: :up
-       }),
-       do: true
+  # Topologically sorts operations using the dependency facts declared by
+  # `AshPostgres.MigrationGenerator.OperationDeps`. Operations are indexed by
+  # their position in the input list; ties (operations with no remaining
+  # dependencies at the same time) are broken by that original index, so
+  # unconstrained operations keep their original relative order (matching
+  # resource declaration order).
+  defp toposort_operations(operations) do
+    count = length(operations)
+    indexed = Enum.with_index(operations)
+
+    provides_index =
+      Enum.reduce(indexed, %{}, fn {op, index}, acc ->
+        op
+        |> OperationDeps.provides()
+        |> Enum.reduce(acc, fn fact, acc ->
+          Map.update(acc, fact, [index], &[index | &1])
+        end)
+      end)
 
-  defp after?(
-         %Operation.AlterAttribute{table: table, schema: schema},
-         %Operation.DropForeignKey{
-           table: table,
-           schema: schema,
-           direction: :down
-         }
-       ),
-       do: false
-
-  defp after?(
-         %Operation.DropForeignKey{
-           table: table,
-           schema: schema,
-           direction: :down
-         },
-         %Operation.AlterAttribute{table: table, schema: schema}
-       ),
-       do: true
+    early_tier_indices =
+      indexed
+      |> Enum.filter(fn {op, _index} -> OperationDeps.early_tier?(op) end)
+      |> Enum.map(&elem(&1, 1))
 
-  defp after?(%Operation.AddAttribute{table: table, schema: schema}, %Operation.CreateTable{
-         table: table,
-         schema: schema
-       }) do
-    true
-  end
+    fact_deps_by_index =
+      Map.new(indexed, fn {op, index} ->
+        {index, op |> OperationDeps.requires() |> Enum.flat_map(&Map.get(provides_index, &1, []))}
+      end)
 
-  defp after?(
-         %Operation.AddAttribute{
-           attribute: %{
-             references: %{table: table, destination_attribute: name}
-           }
-         },
-         %Operation.AddAttribute{table: table, attribute: %{source: name}}
-       ),
-       do: true
+    # An early-tier op (e.g. `RemovePrimaryKey`) can itself have a specific
+    # fact dependency on a *non*-early-tier op (e.g. it requires
+    # `DropForeignKey{direction: :up}` to have run first). Blindly forcing
+    # every non-early-tier op to also come after every early-tier op would
+    # contradict that specific dependency and create a two-op cycle. When an
+    # op is itself something an early-tier op specifically depends on, exempt
+    # it from the blanket "after all early-tier ops" rule — the specific fact
+    # dependency wins.
+    required_by_early_tier =
+      early_tier_indices
+      |> Enum.flat_map(&Map.get(fact_deps_by_index, &1, []))
+      |> MapSet.new()
 
-  defp after?(
-         %Operation.AddAttribute{
-           table: table,
-           schema: schema,
-           attribute: %{
-             primary_key?: false
-           }
-         },
-         %Operation.AddAttribute{schema: schema, table: table, attribute: %{primary_key?: true}}
-       ),
-       do: true
+    dependencies =
+      Map.new(indexed, fn {op, index} ->
+        fact_deps = Map.fetch!(fact_deps_by_index, index)
 
-  defp after?(
-         %Operation.AddAttribute{
-           table: table,
-           schema: schema,
-           attribute: %{
-             primary_key?: true
-           }
-         },
-         %Operation.RemoveAttribute{
-           schema: schema,
-           table: table,
-           attribute: %{primary_key?: true}
-         }
-       ),
-       do: true
+        tier_deps =
+          if OperationDeps.early_tier?(op) || MapSet.member?(required_by_early_tier, index) do
+            []
+          else
+            early_tier_indices
+          end
 
-  defp after?(
-         %Operation.AddAttribute{
-           table: table,
-           schema: schema,
-           attribute: %{
-             primary_key?: true
-           }
-         },
-         %Operation.AlterAttribute{
-           schema: schema,
-           table: table,
-           new_attribute: %{primary_key?: false},
-           old_attribute: %{primary_key?: true}
-         }
-       ),
-       do: true
+        deps =
+          (fact_deps ++ tier_deps)
+          |> Enum.reject(&(&1 == index))
+          |> Enum.uniq()
 
-  defp after?(
-         %Operation.AddAttribute{
-           table: table,
-           schema: schema,
-           attribute: %{
-             primary_key?: true
-           }
-         },
-         %Operation.AlterAttribute{
-           schema: schema,
-           table: table,
-           new_attribute: %{primary_key?: false},
-           old_attribute: %{primary_key?: true}
-         }
-       ),
-       do: true
+        {index, deps}
+      end)
 
-  defp after?(
-         %Operation.RemoveAttribute{
-           schema: schema,
-           table: table,
-           attribute: %{primary_key?: true}
-         },
-         %Operation.AlterAttribute{
-           table: table,
-           schema: schema,
-           new_attribute: %{
-             primary_key?: true
-           },
-           old_attribute: %{
-             primary_key?: false
-           }
-         }
-       ),
-       do: true
+    adjacency =
+      Enum.reduce(dependencies, Map.new(0..(count - 1), &{&1, []}), fn {index, deps}, acc ->
+        Enum.reduce(deps, acc, fn dep_index, acc ->
+          Map.update!(acc, dep_index, &[index | &1])
+        end)
+      end)
 
-  defp after?(
-         %Operation.AlterAttribute{
-           schema: schema,
-           table: table,
-           new_attribute: %{primary_key?: false},
-           old_attribute: %{
-             primary_key?: true
-           }
-         },
-         %Operation.AlterAttribute{
-           table: table,
-           schema: schema,
-           new_attribute: %{
-             primary_key?: true
-           },
-           old_attribute: %{
-             primary_key?: false
-           }
-         }
-       ),
-       do: true
+    in_degrees = Map.new(dependencies, fn {index, deps} -> {index, length(deps)} end)
 
-  defp after?(
-         %Operation.AlterAttribute{
-           schema: schema,
-           table: table,
-           new_attribute: %{primary_key?: false},
-           old_attribute: %{
-             primary_key?: true
-           }
-         },
-         %Operation.AddAttribute{
-           table: table,
-           schema: schema,
-           attribute: %{
-             primary_key?: true
-           }
-         }
-       ),
-       do: false
+    queue =
+      in_degrees
+      |> Enum.filter(fn {_index, in_degree} -> in_degree == 0 end)
+      |> Enum.map(&elem(&1, 0))
+      |> Enum.sort()
 
-  defp after?(
-         %Operation.AlterAttribute{
-           table: table,
-           schema: schema,
-           new_attribute: %{primary_key?: false},
-           old_attribute: %{primary_key?: true}
-         },
-         %Operation.AddAttribute{
-           table: table,
-           schema: schema,
-           attribute: %{
-             primary_key?: true
-           }
-         }
-       ),
-       do: true
+    {ordered_indices, _in_degrees} = toposort_operation_indices(queue, adjacency, in_degrees, [])
 
-  defp after?(
-         %Operation.AlterAttribute{
-           new_attribute: %{
-             references: %{destination_attribute: destination_attribute, table: table}
-           }
-         },
-         %Operation.AddUniqueIndex{identity: %{keys: keys}, table: table}
-       ) do
-    destination_attribute in keys
-  end
+    if length(ordered_indices) == count do
+      operations_tuple = List.to_tuple(operations)
+      Enum.map(ordered_indices, &elem(operations_tuple, &1))
+    else
+      operations_tuple = List.to_tuple(operations)
 
-  defp after?(
-         %Operation.AlterAttribute{
-           old_attribute: %{
-             source: source
-           },
-           table: table,
-           schema: schema
-         },
-         %Operation.RemoveUniqueIndex{identity: %{keys: keys}, table: table, schema: schema}
-       ) do
-    source in List.wrap(keys)
-  end
+      cyclic_operations =
+        0..(count - 1)
+        |> Enum.reject(&(&1 in ordered_indices))
+        |> Enum.map(&elem(operations_tuple, &1))
 
-  defp after?(
-         %Operation.AlterAttribute{
-           new_attribute: %{references: %{table: table, destination_attribute: source}}
-         },
-         %Operation.AlterAttribute{
-           new_attribute: %{
-             source: source
-           },
-           table: table
-         }
-       ) do
-    true
+      raise OperationCycleError, operations: cyclic_operations
+    end
   end
 
-  defp after?(
-         %Operation.AlterAttribute{
-           new_attribute: %{
-             source: source
-           },
-           table: table
-         },
-         %Operation.AlterAttribute{
-           new_attribute: %{references: %{table: table, destination_attribute: source}}
-         }
-       ) do
-    false
+  defp toposort_operation_indices([], _adjacency, in_degrees, acc) do
+    {Enum.reverse(acc), in_degrees}
   end
 
-  defp after?(
-         %Operation.RemoveAttribute{attribute: %{source: source}, table: table},
-         %Operation.AlterAttribute{
-           old_attribute: %{
-             references: %{table: table, destination_attribute: source}
-           }
-         }
-       ),
-       do: true
-
-  defp after?(
-         %Operation.AlterAttribute{
-           new_attribute: %{
-             references: %{table: table, destination_attribute: name}
-           }
-         },
-         %Operation.AddAttribute{table: table, attribute: %{source: name}}
-       ),
-       do: true
-
-  defp after?(%Operation.AddCheckConstraint{table: table, schema: schema}, %Operation.CreateTable{
-         table: table,
-         schema: schema
-       }) do
-    true
-  end
+  defp toposort_operation_indices([index | rest], adjacency, in_degrees, acc) do
+    {in_degrees, newly_available} =
+      Enum.reduce(Map.get(adjacency, index, []), {in_degrees, []}, fn dependent_index,
+                                                                      {in_degrees,
+                                                                       newly_available} ->
+        updated_in_degree = Map.fetch!(in_degrees, dependent_index) - 1
+        in_degrees = Map.put(in_degrees, dependent_index, updated_in_degree)
 
-  defp after?(
-         %Operation.AlterAttribute{new_attribute: %{references: references}, table: table},
-         %{table: table}
-       )
-       when not is_nil(references),
-       do: true
+        if updated_in_degree == 0 do
+          {in_degrees, [dependent_index | newly_available]}
+        else
+          {in_degrees, newly_available}
+        end
+      end)
 
-  defp after?(%Operation.AddCheckConstraint{}, _), do: true
-  defp after?(%Operation.RemoveCheckConstraint{}, _), do: true
+    queue = Enum.sort(rest ++ Enum.uniq(newly_available))
 
-  defp after?(
-         op,
-         %Operation.RenameTable{
-           new_table: table,
-           schema: schema
-         }
-       ) do
-    match?(%{table: ^table, schema: ^schema}, op)
+    toposort_operation_indices(queue, adjacency, in_degrees, [index | acc])
   end
 
-  defp after?(%Operation.RenameTable{}, _), do: false
-
-  defp after?(_, %Operation.DropTable{}), do: true
-  defp after?(%Operation.DropTable{}, _), do: false
-
-  defp after?(
-         %{table: table, schema: schema},
-         %Operation.MoveTableSchema{table: table, schema: schema}
-       ),
-       do: true
-
-  defp after?(%Operation.MoveTableSchema{}, _), do: false
-
-  defp after?(_, _), do: false
-
   defp fetch_operations(snapshots, opts) do
     # Reference diffs need to know when a prefix change is caused by moving
     # the referenced table instead of by changing the foreign key itself.
@@ -2557,14 +2031,26 @@ defmodule AshPostgres.MigrationGenerator do
       |> Enum.reject(fn statement ->
         Enum.any?(old_snapshot.custom_statements, &(&1.name == statement.name))
       end)
-      |> Enum.map(&%Operation.AddCustomStatement{statement: &1, table: snapshot.table})
+      |> Enum.map(
+        &%Operation.AddCustomStatement{
+          statement: &1,
+          table: snapshot.table,
+          schema: snapshot.schema
+        }
+      )
 
     custom_statements_to_remove =
       old_snapshot.custom_statements
       |> Enum.reject(fn old_statement ->
         Enum.any?(snapshot.custom_statements, &(&1.name == old_statement.name))
       end)
-      |> Enum.map(&%Operation.RemoveCustomStatement{statement: &1, table: snapshot.table})
+      |> Enum.map(
+        &%Operation.RemoveCustomStatement{
+          statement: &1,
+          table: snapshot.table,
+          schema: snapshot.schema
+        }
+      )
 
     custom_statements_to_alter =
       snapshot.custom_statements
@@ -2575,8 +2061,16 @@ defmodule AshPostgres.MigrationGenerator do
              (old_statement.code? != statement.code? ||
                 old_statement.up != statement.up || old_statement.down != statement.down) do
           [
-            %Operation.RemoveCustomStatement{statement: old_statement, table: snapshot.table},
-            %Operation.AddCustomStatement{statement: statement, table: snapshot.table}
+            %Operation.RemoveCustomStatement{
+              statement: old_statement,
+              table: snapshot.table,
+              schema: snapshot.schema
+            },
+            %Operation.AddCustomStatement{
+              statement: statement,
+              table: snapshot.table,
+              schema: snapshot.schema
+            }
           ]
         else
           []
@@ -2614,7 +2108,7 @@ defmodule AshPostgres.MigrationGenerator do
             new_index_where = attribute.references && attribute.references[:index_where]
             old_index_where = old_attribute.references && old_attribute.references[:index_where]
 
-            old_index? != new_index? ||
+            (new_index? && old_index? != new_index?) ||
               (new_index? && old_index_where != new_index_where) ||
               (new_index? && multitenancy_changed?)
         end
@@ -2983,7 +2477,11 @@ defmodule AshPostgres.MigrationGenerator do
 
       {[
          must_drop_pkey? &&
-           %Operation.RemovePrimaryKey{schema: snapshot.schema, table: snapshot.table},
+           %Operation.RemovePrimaryKey{
+             schema: snapshot.schema,
+             table: snapshot.table,
+             keys: pkey_names(old_snapshot.attributes)
+           },
          must_drop_pkey? && drop_in_down? &&
            %Operation.RemovePrimaryKeyDown{
              commented?: opts.dont_drop_columns && drop_in_down_commented?,
diff --git a/lib/migration_generator/operation.ex b/lib/migration_generator/operation.ex
index bc7d5197..3d5b3ecb 100644
--- a/lib/migration_generator/operation.ex
+++ b/lib/migration_generator/operation.ex
@@ -157,13 +157,13 @@ defmodule AshPostgres.MigrationGenerator.Operation do
 
   defmodule RenameTable do
     @moduledoc false
-    defstruct [:old_table, :new_table, :schema, :multitenancy, :repo, no_phase: true]
+    defstruct [:old_table, :table, :schema, :multitenancy, :repo, no_phase: true]
 
     import Helper, only: [as_atom: 1]
 
     def up(%{
           old_table: old_table,
-          new_table: new_table,
+          table: new_table,
           schema: schema,
           multitenancy: multitenancy
         }) do
@@ -175,7 +175,7 @@ defmodule AshPostgres.MigrationGenerator.Operation do
 
     def down(%{
           old_table: old_table,
-          new_table: new_table,
+          table: new_table,
           schema: schema,
           multitenancy: multitenancy
         }) do
@@ -1046,7 +1046,7 @@ defmodule AshPostgres.MigrationGenerator.Operation do
 
   defmodule AddCustomStatement do
     @moduledoc false
-    defstruct [:statement, :table, no_phase: true]
+    defstruct [:statement, :table, :schema, no_phase: true]
 
     def up(%{statement: %{up: up, code?: false}}) do
       """
@@ -1075,7 +1075,7 @@ defmodule AshPostgres.MigrationGenerator.Operation do
 
   defmodule RemoveCustomStatement do
     @moduledoc false
-    defstruct [:statement, :table, no_phase: true]
+    defstruct [:statement, :table, :schema, no_phase: true]
 
     def up(%{statement: statement, table: table}) do
       AddCustomStatement.down(%AddCustomStatement{statement: statement, table: table})
@@ -1307,7 +1307,7 @@ defmodule AshPostgres.MigrationGenerator.Operation do
 
   defmodule RemovePrimaryKey do
     @moduledoc false
-    defstruct [:schema, :table, no_phase: true]
+    defstruct [:schema, :table, :keys, no_phase: true]
 
     def up(%{schema: schema, table: table, old_multitenancy: multitenancy}) do
       cond do
diff --git a/lib/migration_generator/operation_cycle_error.ex b/lib/migration_generator/operation_cycle_error.ex
new file mode 100644
index 00000000..d23dc289
--- /dev/null
+++ b/lib/migration_generator/operation_cycle_error.ex
@@ -0,0 +1,44 @@
+# SPDX-FileCopyrightText: 2019 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshPostgres.MigrationGenerator.OperationCycleError do
+  @moduledoc """
+  Raised when the migration operation dependency graph contains a cycle,
+  i.e. two or more operations require facts that only each other provide.
+  """
+  defexception [:operations]
+
+  @impl true
+  def message(%{operations: operations}) do
+    described =
+      operations
+      |> Enum.map(&describe_operation/1)
+      |> Enum.map(&"  - #{&1}")
+      |> Enum.join("\n")
+
+    """
+    Could not determine a valid order for the following migration operations \
+    because they form a dependency cycle:
+
+    #{described}
+
+    If two tables have foreign keys pointing at each other, split the \
+    change into multiple migrations (e.g. create both tables first, add \
+    the foreign keys afterward) or make one of them nullable/deferrable.
+
+    Otherwise, this is most likely a bug in AshPostgres's migration \
+    ordering logic rather than something you can fix in your resources \
+    — please open an issue at \
+    https://github.com/ash-project/ash_postgres/issues with the \
+    resources involved.
+    """
+  end
+
+  defp describe_operation(op) do
+    table = Map.get(op, :table)
+    schema = Map.get(op, :schema)
+
+    "#{inspect(op.__struct__)} on #{inspect(table)}#{if schema, do: " (schema: #{inspect(schema)})", else: ""}"
+  end
+end
diff --git a/lib/migration_generator/operation_deps.ex b/lib/migration_generator/operation_deps.ex
new file mode 100644
index 00000000..6d1310c1
--- /dev/null
+++ b/lib/migration_generator/operation_deps.ex
@@ -0,0 +1,558 @@
+# SPDX-FileCopyrightText: 2019 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshPostgres.MigrationGenerator.OperationDeps do
+  @moduledoc """
+  Computes the dependency graph used to order migration operations.
+
+  Each operation may *provide* facts (things that become true once it runs)
+  and *require* facts (things that must already be true before it can run).
+  `AshPostgres.MigrationGenerator.MigrationGenerator.toposort_operations/1`
+  turns these into a dependency graph and topologically sorts it.
+
+  There is deliberately no symmetric "late tier" counterpart to
+  `early_tier?/1` (a global "runs after everything else" barrier). Adding one
+  creates a real cycle: anything that requires a same-table fact regardless
+  of provider (e.g. `AddCustomStatement` requiring its own table's structure
+  to be ready) would need that "late" op to run first, while the "late" op's
+  barrier would need it to run last — contradictory. Give each operation
+  type the ordering it needs via `requires/1` instead.
+
+  Requiring a fact waits on *every* operation that provides it, not just one
+  — see `toposort_operations/1`'s `provides_index`. That's what makes
+  `:table_structure_ready` work as a catch-all: many operation types provide
+  it, so an op that requires it (currently only `AddCustomStatement`'s
+  own-table requirement) transparently waits for all of that table's
+  structural work, without needing to enumerate every attribute/index/
+  constraint by hand.
+
+  Requiring a fact that nothing in the current batch provides is vacuously
+  satisfied — it adds no dependency edge at all, rather than blocking or
+  raising. This is intentional: e.g. `reference_requirements/1` requires a
+  referenced table/column to be `:column_ready`, but if that table already
+  exists from an earlier migration (so nothing in *this* batch provides the
+  fact), the requirement is trivially met and doesn't hold anything up.
+
+  A generated migration's `down` isn't independently derived — it's built by
+  walking the same operation order in reverse and rendering each operation's
+  `down/1` (see `MigrationGenerator.build_up_and_down/1`). So an ordering
+  that's harmless for `up` can still be wrong: if `RenameAttribute` isn't
+  required to run after a `RemoveCustomIndex` on the same table, `down` ends
+  up recreating that index (`RemoveCustomIndex.down`) *before* undoing the
+  rename, referencing a column name that doesn't exist yet at that point.
+  Postgres itself doesn't need indexes/constraints removed before a rename or
+  a column drop — it tracks index predicates, expression indexes, and
+  same-table constraints internally by column position and updates or drops
+  them automatically (verified directly: a partial unique index's `WHERE`
+  clause and a custom index's expression are both rewritten after `RENAME
+  COLUMN`; a `CHECK` constraint on a dropped column is dropped with it, no
+  `CASCADE` needed). But several facts below exist anyway, purely so the
+  *reversed* `down` sequence stays valid. Only genuinely cross-table
+  dependencies (a foreign key on another table) block a Postgres statement
+  directly, which is what `:column_fk_dropped` is for.
+
+  ## Facts
+
+  Each fact is a 2-tuple `{name, key}`. `key` is `{schema, table}` for a
+  table-scoped fact, or `{schema, table, x}` for a fact scoped to some `x`
+  within that table — see `key/2`, `key/3`. That third element is a column
+  for every fact below except `:custom_index_removed` (an index name) and
+  `:unique_index_created` (a sorted column list); facts are named
+  `table_*`/`column_*`/`index_*` to match what their key actually scopes
+  over, not just its shape.
+
+  Table-scoped (`key = {schema, table}`):
+
+  These first three track "how done is this table", but not as one single
+  chain — `:table_ready` is provided by a disjoint set of operations
+  (`CreateTable`/`RenameTable`/`MoveTableSchema`) from `:table_columns_settled`
+  (`AddAttribute`/`RenameAttribute`/`AlterAttribute`/`RemoveAttribute`), so
+  requiring both together is *not* redundant — neither subsumes the other.
+  Both converge at `:table_structure_ready`, which every structural operation
+  provides:
+
+  - `:table_ready` — the table exists (`CreateTable`/`RenameTable`/
+    `MoveTableSchema`).
+  - `:table_columns_settled` — every attribute add/alter/rename/remove for
+    this table has already run — a conservative margin for consumers whose
+    raw SQL (a filtered unique index's `where`, a custom index's expression,
+    a check constraint's `check:`) might reference a column that's about to
+    be added, altered, renamed, or removed, and can't be parsed to know
+    which columns it actually touches.
+  - `:table_structure_ready` — provided by every structural operation on this
+    table (including `:table_ready`'s and `:table_columns_settled`'s
+    providers); a consumer requires it once to mean "wait for all of this
+    table's structural work".
+
+  Column-scoped (`key = {schema, table, column}`):
+
+  - `:column_ready` — a specific column (by its current name) exists.
+  - `:column_unique_index_removed` — a unique index covering this column has
+    been removed. `RenameAttribute` requires this for the same
+    `down`-validity reason as `:column_custom_index_removed`.
+  - `:column_custom_index_removed` — every `RemoveCustomIndex` whose
+    structured `fields` list includes this column has run. `RenameAttribute`
+    requires this for its own column so `down` stays valid (see above). Only
+    tracks `fields`, not raw `where`/expression text that might reference
+    the column without listing it — a known gap, to be closed later by
+    letting `custom_indexes` declare the columns a raw `where`/expression
+    touches, rather than by conservatively widening this back to table
+    scope.
+  - `:column_check_constraint_removed` — every check constraint covering
+    this column has been removed. `RemoveAttribute`/`RenameAttribute`
+    require this for their own column so `down` stays valid: their `down`
+    recreates/renames the column back, which must happen *before*
+    `RemoveCheckConstraint`'s `down` recreates a constraint that needs it —
+    i.e. `RemoveCheckConstraint` must run first in `up`.
+  - `:column_fk_dropped` — every foreign key on another table that
+    referenced this specific column has been dropped (`direction: :up`).
+    Postgres refuses to drop a unique constraint/primary key that's still
+    referenced by another table's foreign key (verified directly, and not
+    limited to primary keys), so `RemovePrimaryKey`/`RemoveUniqueIndex`
+    require this for each of their own columns.
+  - `:reference_index_removed` — the reference index on this FK column has
+    been dropped (`RemoveReferenceIndex`). `AddReferenceIndex` requires this
+    for its own source column: both derive the same auto-generated index
+    name from their columns, so rewriting an existing reference index
+    (a Remove/Add pair, e.g. when its `index_where` predicate changes) must
+    drop the old index before creating the new one.
+
+  Index-scoped (`key = {schema, table, index}`, same shape as a column-scoped
+  key but the third element is an index name, not a column):
+
+  - `:custom_index_removed` — a specific *named* custom index has been
+    dropped, so a same-named index can be recreated (Postgres index names
+    are unique per schema).
+
+  Column-set-scoped (`key = {schema, table, sorted_columns}`, the third
+  element a sorted list of column atoms):
+
+  - `:unique_index_created` — a unique index covering *exactly* this column
+    set exists (`AddUniqueIndex` for identities, unique `AddCustomIndex`).
+    The set is the columns the rendered index actually covers — including
+    the tenant attribute that attribute-strategy multitenancy prefixes at
+    render time. A foreign key requires this fact for exactly its referenced
+    column set (`destination_attribute` plus `match_with` destinations),
+    matching Postgres's rule that an FK target must be backed by a unique
+    index on exactly the referenced columns — per-column unique indexes
+    don't satisfy it, and correspondingly don't provide this fact.
+  """
+
+  alias AshPostgres.MigrationGenerator.Operation
+
+  @doc "Operation types that must run before every other operation in the batch."
+  def early_tier?(op) do
+    match?(%Operation.DropTable{}, op) ||
+      match?(%Operation.RemoveCustomStatement{}, op) ||
+      match?(%Operation.AlterDeferrability{direction: :down}, op) ||
+      match?(%Operation.RemovePrimaryKey{}, op) ||
+      match?(%Operation.AddPrimaryKeyDown{}, op)
+  end
+
+  @doc "Facts made true once `op` has run."
+  def provides(op) do
+    case op do
+      %Operation.CreateTable{table: table, schema: schema} ->
+        [{:table_ready, key(table, schema)}, {:table_structure_ready, key(table, schema)}]
+
+      %Operation.RenameTable{table: table, schema: schema} ->
+        [{:table_ready, key(table, schema)}, {:table_structure_ready, key(table, schema)}]
+
+      %Operation.MoveTableSchema{table: table, new_schema: schema} ->
+        [{:table_ready, key(table, schema)}, {:table_structure_ready, key(table, schema)}]
+
+      %Operation.AddAttribute{table: table, schema: schema, attribute: attribute} ->
+        [
+          {:column_ready, key(table, schema, attribute.source)},
+          {:table_columns_settled, key(table, schema)},
+          {:table_structure_ready, key(table, schema)}
+        ]
+
+      %Operation.RenameAttribute{
+        table: table,
+        schema: schema,
+        new_attribute: new_attribute
+      } ->
+        [
+          {:column_ready, key(table, schema, new_attribute.source)},
+          {:table_columns_settled, key(table, schema)},
+          {:table_structure_ready, key(table, schema)}
+        ]
+
+      %Operation.AlterAttribute{table: table, schema: schema} ->
+        [
+          {:table_columns_settled, key(table, schema)},
+          {:table_structure_ready, key(table, schema)}
+        ]
+
+      %Operation.RemoveAttribute{table: table, schema: schema} ->
+        [
+          {:table_columns_settled, key(table, schema)},
+          {:table_structure_ready, key(table, schema)}
+        ]
+
+      %Operation.AddUniqueIndex{table: table, schema: schema, identity: identity} = op ->
+        columns =
+          unique_index_column_set(
+            List.wrap(identity.keys),
+            Map.get(identity, :all_tenants?, false),
+            Map.get(op, :multitenancy)
+          )
+
+        [
+          {:table_structure_ready, key(table, schema)},
+          {:unique_index_created, key(table, schema, columns)}
+        ]
+
+      %Operation.RemoveUniqueIndex{table: table, schema: schema, identity: identity} ->
+        keys = List.wrap(identity.keys)
+
+        [{:table_structure_ready, key(table, schema)}] ++
+          Enum.map(keys, &{:column_unique_index_removed, key(table, schema, &1)})
+
+      %Operation.AddCustomIndex{table: table, schema: schema, index: index} = op ->
+        unique_facts =
+          if index.unique do
+            columns =
+              unique_index_column_set(
+                Enum.map(index.fields, &AshPostgres.CustomIndex.column_name/1),
+                Map.get(index, :all_tenants?, false),
+                Map.get(op, :multitenancy)
+              )
+
+            [{:unique_index_created, key(table, schema, columns)}]
+          else
+            []
+          end
+
+        [{:table_structure_ready, key(table, schema)}] ++ unique_facts
+
+      %Operation.RemoveCustomIndex{table: table, schema: schema, index: index} ->
+        base =
+          [{:table_structure_ready, key(table, schema)}] ++
+            Enum.map(index.fields, fn field ->
+              {:column_custom_index_removed,
+               key(table, schema, AshPostgres.CustomIndex.column_name(field))}
+            end)
+
+        # An index without an explicit `name:` gets one auto-derived by
+        # Postgres/Ecto from its columns, so two *different* unnamed indexes
+        # would otherwise collide on the same `nil` fact key and get
+        # spuriously coupled — only track this for explicitly named indexes,
+        # where a real name collision is possible.
+        if index.name do
+          [{:custom_index_removed, key(table, schema, index.name)} | base]
+        else
+          base
+        end
+
+      %Operation.AddReferenceIndex{table: table, schema: schema} ->
+        [{:table_structure_ready, key(table, schema)}]
+
+      %Operation.RemoveReferenceIndex{table: table, schema: schema, source: source} ->
+        [
+          {:table_structure_ready, key(table, schema)},
+          {:reference_index_removed, key(table, schema, source)}
+        ]
+
+      %Operation.AddCheckConstraint{table: table, schema: schema} ->
+        [{:table_structure_ready, key(table, schema)}]
+
+      %Operation.RemoveCheckConstraint{
+        table: table,
+        schema: schema,
+        constraint: constraint,
+        multitenancy: multitenancy
+      } ->
+        cols = check_constraint_columns(constraint, multitenancy)
+
+        [{:table_structure_ready, key(table, schema)}] ++
+          Enum.map(cols, &{:column_check_constraint_removed, key(table, schema, &1)})
+
+      %Operation.DropForeignKey{
+        table: table,
+        schema: schema,
+        direction: :up,
+        attribute: %{
+          references: %{table: dest_table, destination_attribute: dest_column} = reference
+        }
+      } ->
+        dest_schema = Map.get(reference, :schema)
+
+        [
+          {:column_fk_dropped, key(dest_table, dest_schema, dest_column)},
+          {:table_structure_ready, key(table, schema)}
+        ]
+
+      %Operation.DropForeignKey{table: table, schema: schema} ->
+        [{:table_structure_ready, key(table, schema)}]
+
+      %Operation.AlterDeferrability{table: table, schema: schema} ->
+        [{:table_structure_ready, key(table, schema)}]
+
+      %Operation.AddPrimaryKey{table: table, schema: schema} ->
+        [{:table_structure_ready, key(table, schema)}]
+
+      %Operation.AddPrimaryKeyDown{table: table, schema: schema} ->
+        [{:table_structure_ready, key(table, schema)}]
+
+      %Operation.RemovePrimaryKey{table: table, schema: schema} ->
+        [{:table_structure_ready, key(table, schema)}]
+
+      %Operation.RemovePrimaryKeyDown{table: table, schema: schema} ->
+        [{:table_structure_ready, key(table, schema)}]
+
+      _ ->
+        []
+    end
+  end
+
+  @doc "Facts that must already be provided (by some other operation) before `op` can run."
+  def requires(op) do
+    case op do
+      %Operation.AddAttribute{table: table, schema: schema, attribute: attribute} ->
+        [{:table_ready, key(table, schema)}] ++ reference_requirements(attribute)
+
+      %Operation.AlterAttribute{
+        table: table,
+        schema: schema,
+        new_attribute: new_attribute,
+        old_attribute: old_attribute
+      } ->
+        # `old_attribute.source` is normally the current column name, but when
+        # a rename and a property change (type/null/default) land in the same
+        # diff, `old_attribute` here is the *pre-rename* attribute (see
+        # `attribute_operations/4`) while a separate `RenameAttribute` op
+        # handles the physical rename to `new_attribute.source`. Requiring
+        # both names' `column_ready` fact covers both cases: the common case
+        # (no rename, both sources equal, so this is a no-op duplicate) and
+        # the rename+alter case (only `new_attribute.source` has a provider —
+        # the `RenameAttribute` op — so the alter correctly waits for it).
+        [
+          {:table_ready, key(table, schema)},
+          {:column_ready, key(table, schema, old_attribute.source)},
+          {:column_ready, key(table, schema, new_attribute.source)}
+        ] ++ reference_requirements(new_attribute)
+
+      %Operation.RenameAttribute{table: table, schema: schema, old_attribute: old_attribute} ->
+        [
+          {:table_ready, key(table, schema)},
+          {:column_ready, key(table, schema, old_attribute.source)},
+          {:column_unique_index_removed, key(table, schema, old_attribute.source)},
+          {:column_check_constraint_removed, key(table, schema, old_attribute.source)},
+          # This table's `down` sequence is this same operation list reversed
+          # (see moduledoc), so a RemoveCustomIndex covering this column must
+          # run first in `up` — otherwise its `down` (recreating the index)
+          # would run before this rename's `down` undoes the rename,
+          # referencing a column name that doesn't exist yet. Only tracks
+          # the index's structured `fields` list, not raw `where`/expression
+          # text that might reference this column without listing it —
+          # accepted gap for now (see moduledoc note on custom_index_removed).
+          {:column_custom_index_removed, key(table, schema, old_attribute.source)}
+        ]
+
+      %Operation.RemoveAttribute{table: table, schema: schema, attribute: attribute} ->
+        [
+          {:table_ready, key(table, schema)},
+          {:column_check_constraint_removed, key(table, schema, attribute.source)}
+        ]
+
+      %Operation.AddUniqueIndex{
+        table: table,
+        schema: schema,
+        identity: identity,
+        insert_after_attribute_source: source
+      } ->
+        base = [{:table_ready, key(table, schema)}]
+
+        # `where`/`base_filter` on a unique index (e.g. a soft-delete scoped
+        # identity filtering on `archived_at IS NULL`) can reference columns
+        # that aren't otherwise declared as this index's keys, and there's no
+        # structured way to know which ones without parsing raw SQL. Requiring
+        # every column op on the table (same margin `AddCustomIndex` already
+        # takes) is the safe choice — but only when such a filter actually
+        # exists: unconditionally requiring this would also pull in a
+        # self-referencing attribute whose own FK targets *this* index (e.g.
+        # a self-referential `belongs_to`), creating a genuine two-way
+        # dependency cycle between that attribute and this index.
+        base =
+          if Map.get(identity, :where) || Map.get(identity, :base_filter) do
+            [{:table_columns_settled, key(table, schema)} | base]
+          else
+            base
+          end
+
+        if source do
+          [{:column_ready, key(table, schema, source)} | base]
+        else
+          base
+        end
+
+      %Operation.RemoveUniqueIndex{table: table, schema: schema, identity: identity} ->
+        keys = List.wrap(identity.keys)
+
+        # Postgres refuses to drop a unique constraint/index that's still
+        # referenced by another table's foreign key (the same restriction
+        # `RemovePrimaryKey` has, just not limited to the primary key —
+        # verified directly: dropping a non-PK UNIQUE constraint still
+        # referenced by an FK raises "other objects depend on it").
+        [{:table_ready, key(table, schema)}] ++
+          Enum.map(keys, &{:column_fk_dropped, key(table, schema, &1)})
+
+      %Operation.AddCustomIndex{table: table, schema: schema, index: index} ->
+        # When the index has no raw `where` and every field is a plain column
+        # name, we know exactly which columns it touches — require just those
+        # columns' existence instead of the whole table being settled. Raw SQL
+        # (a `where`, an expression field like "lower(email)") keeps the
+        # conservative table-wide margin. The precision matters beyond
+        # performance: a *unique* custom index provides
+        # `unique_index_created` (see `provides/1`) so that FKs
+        # targeting it run after it — with a table-wide requirement here, a
+        # same-table FK alter would wait on this index while this index waits
+        # on `table_columns_settled` provided by that same alter: a cycle.
+        column_requirements =
+          case simple_column_fields(index) do
+            {:ok, columns} ->
+              Enum.map(columns, &{:column_ready, key(table, schema, &1)})
+
+            :error ->
+              [{:table_columns_settled, key(table, schema)}]
+          end
+
+        base = [{:table_ready, key(table, schema)} | column_requirements]
+
+        # a `RemoveCustomIndex` sharing this index's *explicit* name must run
+        # first — Postgres can't create an index under a name that's still in
+        # use. See the matching guard in `provides/1`.
+        if index.name do
+          [{:custom_index_removed, key(table, schema, index.name)} | base]
+        else
+          base
+        end
+
+      %Operation.AddReferenceIndex{table: table, schema: schema, source: source} ->
+        # a `RemoveReferenceIndex` on the same source column must run first —
+        # both derive the same auto-generated index name from their columns,
+        # so Postgres can't create the new index while the old one exists.
+        [
+          {:table_ready, key(table, schema)},
+          {:table_columns_settled, key(table, schema)},
+          {:reference_index_removed, key(table, schema, source)}
+        ]
+
+      %Operation.AddCheckConstraint{
+        table: table,
+        schema: schema,
+        constraint: constraint,
+        multitenancy: multitenancy
+      } ->
+        cols = check_constraint_columns(constraint, multitenancy)
+
+        [{:table_ready, key(table, schema)}, {:table_columns_settled, key(table, schema)}] ++
+          Enum.map(cols, &{:column_ready, key(table, schema, &1)})
+
+      %Operation.RemoveCheckConstraint{table: table, schema: schema} ->
+        [{:table_ready, key(table, schema)}]
+
+      %Operation.DropForeignKey{table: table, schema: schema} ->
+        [{:table_ready, key(table, schema)}]
+
+      %Operation.RemovePrimaryKey{table: table, schema: schema, keys: keys} ->
+        Enum.map(List.wrap(keys), &{:column_fk_dropped, key(table, schema, &1)})
+
+      %Operation.AddCustomStatement{table: own_table, schema: schema} ->
+        [{:table_structure_ready, key(own_table, schema)}]
+
+      _ ->
+        []
+    end
+  end
+
+  defp reference_requirements(%{
+         references: %{table: table, destination_attribute: column} = reference
+       }) do
+    schema = Map.get(reference, :schema)
+
+    # A composite foreign key (`with:` via match_with) references the
+    # destination column plus the match_with destination columns. Postgres
+    # requires one unique index covering exactly that column set, so the FK
+    # must run after each referenced column exists and after the covering
+    # unique index (when either is created in this same batch).
+    destination_columns =
+      [column | Map.values(Map.get(reference, :match_with) || %{})]
+      |> Enum.map(&normalize_column/1)
+
+    column_set = destination_columns |> Enum.uniq() |> Enum.sort()
+
+    Enum.map(destination_columns, &{:column_ready, key(table, schema, &1)}) ++
+      [{:unique_index_created, key(table, schema, column_set)}]
+  end
+
+  defp reference_requirements(_), do: []
+
+  # The columns the rendered unique index actually covers: attribute-strategy
+  # multitenancy prefixes the tenant attribute at render time (see
+  # `Operation.Helper.index_keys/3`), so the fact key must be built from the
+  # same column set for it to line up with what a composite FK references.
+  defp unique_index_column_set(keys, all_tenants?, multitenancy) do
+    keys
+    |> Operation.Helper.index_keys(all_tenants?, multitenancy || %{strategy: nil})
+    |> Enum.map(&normalize_column/1)
+    |> Enum.uniq()
+    |> Enum.sort()
+  end
+
+  defp normalize_column(column) when is_atom(column), do: column
+  # sobelow_skip ["DOS.StringToAtom"]
+  defp normalize_column(column) when is_binary(column), do: String.to_atom(column)
+
+  # {:ok, columns} when the index provably touches exactly `columns` (no raw
+  # `where`, every field and included column a plain column name); :error when
+  # any raw SQL means we can't know.
+  defp simple_column_fields(index) do
+    fields = Map.get(index, :fields)
+
+    if Map.get(index, :where) || is_nil(fields) do
+      :error
+    else
+      columns =
+        Enum.map(fields, &AshPostgres.CustomIndex.column_name/1) ++
+          List.wrap(Map.get(index, :include))
+
+      if Enum.all?(columns, fn column ->
+           column |> to_string() |> String.match?(~r/^[a-zA-Z0-9_]+$/)
+         end) do
+        {:ok, Enum.map(columns, &normalize_column/1)}
+      else
+        :error
+      end
+    end
+  end
+
+  defp check_constraint_columns(constraint, multitenancy) do
+    cols = List.wrap(Map.get(constraint, :attribute))
+
+    if multitenancy && Map.get(multitenancy, :attribute) do
+      Enum.uniq([multitenancy.attribute | cols])
+    else
+      cols
+    end
+  end
+
+  # `op.schema` (a resource's own postgres schema option) defaults to `nil` for
+  # the default schema, but `attribute.references.schema` (the *destination*
+  # schema recorded on a foreign key) is instead loaded from the snapshot as
+  # the literal string `"public"` (see `load_attribute/2` in
+  # migration_generator.ex, which only `Map.put_new/3`s a default and leaves
+  # an explicit `"public"` alone). Without normalizing these to the same
+  # value, a same-table fact key built from `op.schema` (`nil`) would never
+  # match a cross-table reference fact key built from `reference.schema`
+  # (`"public"`) even though they mean the same schema, silently dropping the
+  # ordering dependency.
+  defp key(table, schema), do: {normalize_schema(schema), table}
+  defp key(table, schema, col), do: {normalize_schema(schema), table, col}
+
+  defp normalize_schema(nil), do: "public"
+  defp normalize_schema(schema), do: schema
+end
diff --git a/test/migration_generator/operation_deps_test.exs b/test/migration_generator/operation_deps_test.exs
new file mode 100644
index 00000000..3c60294b
--- /dev/null
+++ b/test/migration_generator/operation_deps_test.exs
@@ -0,0 +1,609 @@
+# SPDX-FileCopyrightText: 2019 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshPostgres.MigrationGenerator.OperationDepsTest do
+  @moduledoc """
+  Unit tests for `AshPostgres.MigrationGenerator.OperationDeps`, the
+  dependency-graph model used to order migration operations.
+  """
+  use ExUnit.Case, async: true
+
+  alias AshPostgres.MigrationGenerator.Operation
+  alias AshPostgres.MigrationGenerator.OperationDeps
+
+  describe "table existence" do
+    test "table_ready from CreateTable, RenameTable, or MoveTableSchema each satisfy an AddAttribute on that table" do
+      consumer = %Operation.AddAttribute{
+        table: "posts",
+        schema: nil,
+        attribute: %{source: :title, primary_key?: false}
+      }
+
+      [required_fact] =
+        OperationDeps.requires(consumer) |> Enum.filter(&match?({:table_ready, _}, &1))
+
+      for provider <- [
+            %Operation.CreateTable{table: "posts", schema: nil},
+            %Operation.RenameTable{old_table: "old_posts", table: "posts", schema: nil},
+            %Operation.MoveTableSchema{table: "posts", old_schema: "x", new_schema: nil}
+          ] do
+        assert required_fact in OperationDeps.provides(provider),
+               "expected #{inspect(provider.__struct__)} to satisfy AddAttribute's table_ready requirement"
+      end
+    end
+  end
+
+  describe "same-table column existence" do
+    test "AlterAttribute's column_ready requirement is satisfied by the AddAttribute that created the column" do
+      add = %Operation.AddAttribute{
+        table: "posts",
+        schema: nil,
+        attribute: %{source: :title, primary_key?: false}
+      }
+
+      alter = %Operation.AlterAttribute{
+        table: "posts",
+        schema: nil,
+        old_attribute: %{source: :title},
+        new_attribute: %{source: :title}
+      }
+
+      [column_ready_fact] =
+        OperationDeps.provides(add) |> Enum.filter(&match?({:column_ready, _}, &1))
+
+      assert column_ready_fact in OperationDeps.requires(alter)
+    end
+  end
+
+  describe "cross-table structural FK" do
+    test "an AddAttribute with a structural reference is satisfied by the referenced table's column and unique index" do
+      referenced_column = %Operation.AddAttribute{
+        table: "posts",
+        schema: nil,
+        attribute: %{source: :id, primary_key?: true}
+      }
+
+      referenced_index = %Operation.AddUniqueIndex{
+        table: "posts",
+        schema: nil,
+        identity: %{keys: [:id], where: nil, base_filter: nil}
+      }
+
+      referencing_attribute = %Operation.AddAttribute{
+        table: "comments",
+        schema: nil,
+        attribute: %{
+          source: :post_id,
+          primary_key?: false,
+          references: %{table: "posts", destination_attribute: :id, schema: "public"}
+        }
+      }
+
+      requires = OperationDeps.requires(referencing_attribute)
+
+      [column_fact] =
+        OperationDeps.provides(referenced_column) |> Enum.filter(&match?({:column_ready, _}, &1))
+
+      [index_fact] =
+        OperationDeps.provides(referenced_index)
+        |> Enum.filter(&match?({:unique_index_created, _}, &1))
+
+      assert index_fact == {:unique_index_created, {"public", "posts", [:id]}}
+      assert column_fact in requires
+      assert index_fact in requires
+    end
+
+    test "a composite FK (match_with) requires a unique index covering exactly its referenced column set" do
+      # e.g. `reference :dept, match_with: [org_id: :org_id]` produces a
+      # composite FK on dept (id, org_id); Postgres requires one unique index
+      # covering exactly those columns, which here is a custom index rather
+      # than an identity (identities mix in base_filter, making the index
+      # partial and unusable as an FK target).
+      referenced_index = %Operation.AddCustomIndex{
+        table: "dept",
+        schema: nil,
+        index: %{name: "dept_org_id_id_index", unique: true, fields: ["org_id", "id"]},
+        base_filter: nil,
+        multitenancy: %{attribute: nil, strategy: nil, global: nil}
+      }
+
+      referencing_attribute = %Operation.AddAttribute{
+        table: "customer",
+        schema: nil,
+        attribute: %{
+          source: :dept_id,
+          primary_key?: false,
+          references: %{
+            table: "dept",
+            destination_attribute: :id,
+            schema: "public",
+            match_with: %{org_id: :org_id}
+          }
+        }
+      }
+
+      requires = OperationDeps.requires(referencing_attribute)
+
+      # provider side: string field names normalize into one sorted atom set
+      [index_fact] =
+        OperationDeps.provides(referenced_index)
+        |> Enum.filter(&match?({:unique_index_created, _}, &1))
+
+      assert index_fact == {:unique_index_created, {"public", "dept", [:id, :org_id]}}
+
+      # consumer side: the FK waits for the covering index and the extra column
+      assert index_fact in requires
+      assert {:column_ready, {"public", "dept", :org_id}} in requires
+    end
+
+    test "separate single-column unique indexes do not satisfy a composite FK's covering-index requirement" do
+      # Postgres requires ONE unique index on exactly (id, org_id); a unique
+      # index on (id) plus another on (org_id) does not qualify, and
+      # correspondingly neither provides the column-set fact the FK requires.
+      id_index = %Operation.AddCustomIndex{
+        table: "dept",
+        schema: nil,
+        index: %{name: nil, unique: true, fields: ["id"]},
+        base_filter: nil,
+        multitenancy: %{attribute: nil, strategy: nil, global: nil}
+      }
+
+      org_id_index = %Operation.AddCustomIndex{
+        table: "dept",
+        schema: nil,
+        index: %{name: nil, unique: true, fields: ["org_id"]},
+        base_filter: nil,
+        multitenancy: %{attribute: nil, strategy: nil, global: nil}
+      }
+
+      referencing_attribute = %Operation.AddAttribute{
+        table: "customer",
+        schema: nil,
+        attribute: %{
+          source: :dept_id,
+          primary_key?: false,
+          references: %{
+            table: "dept",
+            destination_attribute: :id,
+            schema: "public",
+            match_with: %{org_id: :org_id}
+          }
+        }
+      }
+
+      [required_index_fact] =
+        OperationDeps.requires(referencing_attribute)
+        |> Enum.filter(&match?({:unique_index_created, _}, &1))
+
+      refute required_index_fact in OperationDeps.provides(id_index)
+      refute required_index_fact in OperationDeps.provides(org_id_index)
+    end
+
+    test "the provided column set includes the tenant attribute that multitenancy prefixes at render time" do
+      # The rendered index is (org_id, secondary_id), not (secondary_id) —
+      # see Operation.Helper.index_keys/3 — so the fact must say so, or a
+      # composite FK targeting (org_id, secondary_id) would never link to it.
+      op = %Operation.AddUniqueIndex{
+        table: "users",
+        schema: nil,
+        identity: %{keys: [:secondary_id], where: nil, base_filter: nil},
+        multitenancy: %{strategy: :attribute, attribute: :org_id, global: false}
+      }
+
+      assert {:unique_index_created, {"public", "users", [:org_id, :secondary_id]}} in OperationDeps.provides(
+               op
+             )
+    end
+
+    test "a non-unique custom index does not provide unique_index_created" do
+      index_op = %Operation.AddCustomIndex{
+        table: "dept",
+        schema: nil,
+        index: %{name: nil, unique: false, fields: ["org_id"]},
+        base_filter: nil,
+        multitenancy: %{attribute: nil, strategy: nil, global: nil}
+      }
+
+      refute Enum.any?(
+               OperationDeps.provides(index_op),
+               &match?({:unique_index_created, _}, &1)
+             )
+    end
+
+    test "reference.schema (\"public\" string) and a table's own nil schema normalize to the same fact key" do
+      # attribute.references.schema is loaded from the snapshot as an explicit
+      # "public" string, while a resource's own `schema` option defaults to
+      # `nil` for the default schema. Both must resolve to the same fact key,
+      # or a real cross-table FK dependency silently disappears.
+      provider_op = %Operation.AddAttribute{
+        table: "posts",
+        schema: nil,
+        attribute: %{source: :id, primary_key?: true}
+      }
+
+      consumer_op = %Operation.AddAttribute{
+        table: "comments",
+        schema: nil,
+        attribute: %{
+          source: :post_id,
+          primary_key?: false,
+          references: %{table: "posts", destination_attribute: :id, schema: "public"}
+        }
+      }
+
+      [provided_fact] =
+        OperationDeps.provides(provider_op) |> Enum.filter(&match?({:column_ready, _}, &1))
+
+      assert provided_fact in OperationDeps.requires(consumer_op)
+    end
+  end
+
+  describe "unique index where/base_filter columns" do
+    test "a filtered identity requires table_columns_settled (can't know which columns the raw SQL filter touches)" do
+      filtered = %Operation.AddUniqueIndex{
+        table: "posts",
+        schema: nil,
+        identity: %{keys: [:seq], where: "archived_at IS NULL", base_filter: nil},
+        insert_after_attribute_source: nil
+      }
+
+      plain = %Operation.AddUniqueIndex{
+        table: "posts",
+        schema: nil,
+        identity: %{keys: [:seq], where: nil, base_filter: nil},
+        insert_after_attribute_source: nil
+      }
+
+      assert {:table_columns_settled, {"public", "posts"}} in OperationDeps.requires(filtered)
+      refute {:table_columns_settled, {"public", "posts"}} in OperationDeps.requires(plain)
+    end
+
+    test "a plain (unfiltered) identity referencing a self-FK attribute does not require table_columns_settled (regression: issue #236 self-reference cycle)" do
+      # A self-referencing belongs_to's FK attribute (e.g. `follows`) can be
+      # the *last* AddAttribute for the table, so requiring "any attribute
+      # added" here would create a cycle against that attribute's own
+      # dependency on this very index. Only filtered identities take the
+      # broader (and unavoidably imprecise) margin.
+      op = %Operation.AddUniqueIndex{
+        table: "template_phase",
+        schema: nil,
+        identity: %{keys: [:id], where: nil, base_filter: nil},
+        insert_after_attribute_source: nil
+      }
+
+      refute {:table_columns_settled, {"public", "template_phase"}} in OperationDeps.requires(op)
+    end
+  end
+
+  describe "custom index name collisions" do
+    test "RemoveCustomIndex and AddCustomIndex sharing an explicit name are linked" do
+      remove = %Operation.RemoveCustomIndex{
+        table: "users",
+        schema: nil,
+        index: %{name: "users_active_name_index", fields: []},
+        base_filter: nil,
+        multitenancy: %{attribute: nil, strategy: nil, global: nil}
+      }
+
+      add = %Operation.AddCustomIndex{
+        table: "users",
+        schema: nil,
+        index: %{name: "users_active_name_index"},
+        base_filter: nil,
+        multitenancy: %{attribute: nil, strategy: nil, global: nil}
+      }
+
+      [fact] =
+        OperationDeps.provides(remove) |> Enum.filter(&match?({:custom_index_removed, _}, &1))
+
+      assert fact in OperationDeps.requires(add)
+    end
+
+    test "two differently-shaped unnamed indexes are NOT spuriously linked by a shared nil name" do
+      # Regression: an auto-named (nil `name:`) index previously collapsed to
+      # the same `{schema, table, nil}` fact key as any other unnamed index on
+      # the same table, incorrectly coupling unrelated index changes.
+      remove = %Operation.RemoveCustomIndex{
+        table: "posts",
+        schema: nil,
+        index: %{name: nil, fields: []},
+        base_filter: nil,
+        multitenancy: %{attribute: nil, strategy: nil, global: nil}
+      }
+
+      add = %Operation.AddCustomIndex{
+        table: "posts",
+        schema: nil,
+        index: %{name: nil},
+        base_filter: nil,
+        multitenancy: %{attribute: nil, strategy: nil, global: nil}
+      }
+
+      refute Enum.any?(
+               OperationDeps.provides(remove),
+               &match?({:custom_index_removed, _}, &1)
+             )
+
+      refute Enum.any?(OperationDeps.requires(add), &match?({:custom_index_removed, _}, &1))
+    end
+
+    test "RemoveReferenceIndex and AddReferenceIndex on the same source column are linked" do
+      # Both derive the same auto-generated index name from their columns, so
+      # the add can't run while the old index still exists (e.g. an index_where
+      # rewrite of an existing reference index).
+      remove = %Operation.RemoveReferenceIndex{
+        table: "posts",
+        schema: nil,
+        source: :author_id,
+        multitenancy: %{attribute: nil, strategy: nil, global: nil},
+        old_multitenancy: %{attribute: nil, strategy: nil, global: nil}
+      }
+
+      add = %Operation.AddReferenceIndex{
+        table: "posts",
+        schema: nil,
+        source: :author_id,
+        where: "author_id IS NOT NULL",
+        multitenancy: %{attribute: nil, strategy: nil, global: nil}
+      }
+
+      [fact] =
+        OperationDeps.provides(remove)
+        |> Enum.filter(&match?({:reference_index_removed, _}, &1))
+
+      assert fact in OperationDeps.requires(add)
+    end
+
+    test "AddReferenceIndex is NOT linked to a RemoveReferenceIndex on a different source column" do
+      remove = %Operation.RemoveReferenceIndex{
+        table: "posts",
+        schema: nil,
+        source: :editor_id,
+        multitenancy: %{attribute: nil, strategy: nil, global: nil},
+        old_multitenancy: %{attribute: nil, strategy: nil, global: nil}
+      }
+
+      add = %Operation.AddReferenceIndex{
+        table: "posts",
+        schema: nil,
+        source: :author_id,
+        where: nil,
+        multitenancy: %{attribute: nil, strategy: nil, global: nil}
+      }
+
+      [fact] =
+        OperationDeps.provides(remove)
+        |> Enum.filter(&match?({:reference_index_removed, _}, &1))
+
+      refute fact in OperationDeps.requires(add)
+    end
+  end
+
+  describe "down-sequence validity for renames" do
+    # A generated migration's `down` is the same operation order reversed,
+    # rendering each operation's `down/1` (see the moduledoc). So even though
+    # Postgres itself doesn't need an index/constraint removed before a
+    # rename, RenameAttribute still has to run after these — otherwise
+    # `down` would recreate the old index/unique-constraint (referencing the
+    # old column name) before the rename's `down` restores that name.
+
+    test "RenameAttribute requires column_custom_index_removed, satisfied by a RemoveCustomIndex whose fields include that column" do
+      remove = %Operation.RemoveCustomIndex{
+        table: "posts",
+        schema: nil,
+        index: %{name: nil, fields: [:title]},
+        base_filter: nil,
+        multitenancy: %{attribute: nil, strategy: nil, global: nil}
+      }
+
+      rename = %Operation.RenameAttribute{
+        table: "posts",
+        schema: nil,
+        old_attribute: %{source: :title},
+        new_attribute: %{source: :title_short}
+      }
+
+      [fact] =
+        OperationDeps.provides(remove)
+        |> Enum.filter(&match?({:column_custom_index_removed, _}, &1))
+
+      assert fact in OperationDeps.requires(rename)
+    end
+
+    test "RenameAttribute is NOT satisfied by a RemoveCustomIndex covering a different column" do
+      remove = %Operation.RemoveCustomIndex{
+        table: "posts",
+        schema: nil,
+        index: %{name: nil, fields: [:body]},
+        base_filter: nil,
+        multitenancy: %{attribute: nil, strategy: nil, global: nil}
+      }
+
+      rename = %Operation.RenameAttribute{
+        table: "posts",
+        schema: nil,
+        old_attribute: %{source: :title},
+        new_attribute: %{source: :title_short}
+      }
+
+      [fact] =
+        OperationDeps.provides(remove)
+        |> Enum.filter(&match?({:column_custom_index_removed, _}, &1))
+
+      refute fact in OperationDeps.requires(rename)
+    end
+
+    test "RenameAttribute requires column_unique_index_removed, satisfied by the RemoveUniqueIndex that covered the old column name" do
+      remove_index = %Operation.RemoveUniqueIndex{
+        table: "posts",
+        schema: nil,
+        identity: %{keys: [:title]}
+      }
+
+      rename = %Operation.RenameAttribute{
+        table: "posts",
+        schema: nil,
+        old_attribute: %{source: :title},
+        new_attribute: %{source: :title_short}
+      }
+
+      [fact] =
+        OperationDeps.provides(remove_index)
+        |> Enum.filter(&match?({:column_unique_index_removed, _}, &1))
+
+      assert fact in OperationDeps.requires(rename)
+    end
+
+    test "RemoveAttribute requires column_check_constraint_removed, satisfied by the RemoveCheckConstraint that covered it" do
+      # Confirmed by a real generated migration (see migration_generator_test.exs
+      # "check constraint and column removed together"): RemoveCheckConstraint
+      # must run before RemoveAttribute in `up`, or `down` tries to recreate
+      # the constraint (RemoveCheckConstraint.down) before the column it
+      # covers exists again (RemoveAttribute.down hasn't run yet in the
+      # reversed sequence) — a real Postgres "column does not exist" error.
+      remove_constraint = %Operation.RemoveCheckConstraint{
+        table: "posts",
+        schema: nil,
+        constraint: %{attribute: [:title]},
+        multitenancy: nil
+      }
+
+      remove_attribute = %Operation.RemoveAttribute{
+        table: "posts",
+        schema: nil,
+        attribute: %{source: :title}
+      }
+
+      [fact] =
+        OperationDeps.provides(remove_constraint)
+        |> Enum.filter(&match?({:column_check_constraint_removed, _}, &1))
+
+      assert fact in OperationDeps.requires(remove_attribute)
+    end
+  end
+
+  describe "custom_statements" do
+    test "AddCustomStatement's own-table requirement is satisfied by a CreateTable for that same table" do
+      create = %Operation.CreateTable{table: "widget", schema: nil}
+
+      statement = %Operation.AddCustomStatement{
+        table: "widget",
+        schema: nil,
+        statement: %{name: :some_statement, up: "", down: "", code?: false}
+      }
+
+      [fact] =
+        OperationDeps.provides(create) |> Enum.filter(&match?({:table_structure_ready, _}, &1))
+
+      assert fact in OperationDeps.requires(statement)
+    end
+  end
+
+  describe "early tier" do
+    test "DropTable, RemoveCustomStatement, and the down-direction deferrability op are early tier" do
+      assert OperationDeps.early_tier?(%Operation.DropTable{table: "posts", schema: nil})
+
+      assert OperationDeps.early_tier?(%Operation.RemoveCustomStatement{
+               table: "posts",
+               statement: %{name: :x, up: "", down: "", code?: false}
+             })
+
+      assert OperationDeps.early_tier?(%Operation.AlterDeferrability{
+               table: "posts",
+               schema: nil,
+               references: %{},
+               direction: :down
+             })
+
+      refute OperationDeps.early_tier?(%Operation.AlterDeferrability{
+               table: "posts",
+               schema: nil,
+               references: %{},
+               direction: :up
+             })
+    end
+
+    test "RemovePrimaryKey and AddPrimaryKeyDown are early tier; RemovePrimaryKeyDown and AddPrimaryKey are not" do
+      assert OperationDeps.early_tier?(%Operation.RemovePrimaryKey{table: "posts", schema: nil})
+
+      assert OperationDeps.early_tier?(%Operation.AddPrimaryKeyDown{
+               table: "posts",
+               schema: nil,
+               keys: [:id],
+               remove_old?: false
+             })
+
+      refute OperationDeps.early_tier?(%Operation.RemovePrimaryKeyDown{
+               table: "posts",
+               schema: nil
+             })
+
+      refute OperationDeps.early_tier?(%Operation.AddPrimaryKey{
+               table: "posts",
+               schema: nil,
+               keys: [:id]
+             })
+    end
+
+    test "RemovePrimaryKey requires column_fk_dropped for each of its own PK columns, satisfied only by a DropForeignKey{direction: :up} targeting that exact column" do
+      remove_pk = %Operation.RemovePrimaryKey{table: "posts", schema: nil, keys: [:id]}
+
+      # "comments" owns the FK column, but it targets "posts.id" — the fact
+      # is scoped by the *referenced* table+column (Postgres blocks dropping
+      # a PK while another table's FK still points at it), not the table
+      # owning the FK.
+      drop_fk_targeting_posts_id = %Operation.DropForeignKey{
+        table: "comments",
+        schema: nil,
+        attribute: %{references: %{table: "posts", destination_attribute: :id, schema: nil}},
+        direction: :up
+      }
+
+      drop_fk_targeting_other_column = %Operation.DropForeignKey{
+        table: "comments",
+        schema: nil,
+        attribute: %{references: %{table: "posts", destination_attribute: :slug, schema: nil}},
+        direction: :up
+      }
+
+      [fact] =
+        OperationDeps.provides(drop_fk_targeting_posts_id)
+        |> Enum.filter(&match?({:column_fk_dropped, _}, &1))
+
+      assert fact in OperationDeps.requires(remove_pk)
+
+      refute Enum.any?(
+               OperationDeps.provides(drop_fk_targeting_other_column),
+               &(&1 == fact)
+             )
+    end
+
+    test "RemoveUniqueIndex requires column_fk_dropped for its own columns too (not just RemovePrimaryKey)" do
+      # Regression: Postgres refuses to drop *any* unique constraint/index
+      # still referenced by another table's foreign key, not just a primary
+      # key (verified directly against Postgres) — a `belongs_to` with
+      # `destination_attribute` can target a non-PK unique identity via
+      # plain Ash DSL, so this is a reachable real scenario, not just a
+      # custom_statements edge case.
+      remove_index = %Operation.RemoveUniqueIndex{
+        table: "parents",
+        schema: nil,
+        identity: %{keys: [:code]}
+      }
+
+      drop_fk_targeting_code = %Operation.DropForeignKey{
+        table: "children",
+        schema: nil,
+        attribute: %{references: %{table: "parents", destination_attribute: :code, schema: nil}},
+        direction: :up
+      }
+
+      [fact] =
+        OperationDeps.provides(drop_fk_targeting_code)
+        |> Enum.filter(&match?({:column_fk_dropped, _}, &1))
+
+      assert fact in OperationDeps.requires(remove_index)
+    end
+  end
+end
diff --git a/test/migration_generator_test.exs b/test/migration_generator_test.exs
index f52f11bd..5e3bc352 100644
--- a/test/migration_generator_test.exs
+++ b/test/migration_generator_test.exs
@@ -943,7 +943,9 @@ defmodule AshPostgres.MigrationGeneratorTest do
 
       [up_side, down_side] = String.split(contents, "def down", parts: 2)
 
-      up_side_parts = String.split(up_side, "\n", trim: true)
+      up_side_parts =
+        String.split(up_side, "\n", trim: true)
+        |> Enum.map(&String.trim/1)
 
       assert Enum.find_index(up_side_parts, fn x ->
                x == "rename table(:posts), :title, to: :title_short"
@@ -952,7 +954,9 @@ defmodule AshPostgres.MigrationGeneratorTest do
                  x == "create index(:posts, [:title_short])"
                end)
 
-      down_side_parts = String.split(down_side, "\n", trim: true)
+      down_side_parts =
+        String.split(down_side, "\n", trim: true)
+        |> Enum.map(&String.trim/1)
 
       assert Enum.find_index(down_side_parts, fn x ->
                x == "rename table(:posts), :title_short, to: :title"
@@ -961,6 +965,152 @@ defmodule AshPostgres.MigrationGeneratorTest do
     end
   end
 
+  describe "check constraint and column removed together" do
+    setup %{snapshot_path: snapshot_path, migration_path: migration_path} do
+      :ok
+
+      defposts do
+        attributes do
+          uuid_primary_key(:id)
+          attribute(:price, :integer, public?: true)
+        end
+
+        postgres do
+          check_constraints do
+            check_constraint(:price, "price_must_be_positive", check: ~S["price" > 0])
+          end
+        end
+      end
+
+      defdomain([Post])
+
+      AshPostgres.MigrationGenerator.generate(Domain,
+        snapshot_path: snapshot_path,
+        migration_path: migration_path,
+        quiet: true,
+        format: false,
+        auto_name: true
+      )
+    end
+
+    test "the constraint is dropped before the column, so `down` recreates the column before the constraint",
+         %{snapshot_path: snapshot_path, migration_path: migration_path} do
+      defposts do
+        attributes do
+          uuid_primary_key(:id)
+        end
+      end
+
+      defdomain([Post])
+
+      send(self(), {:mix_shell_input, :yes?, true})
+
+      AshPostgres.MigrationGenerator.generate(Domain,
+        snapshot_path: snapshot_path,
+        migration_path: migration_path,
+        quiet: true,
+        format: false,
+        auto_name: true
+      )
+
+      assert [_file1, file2] =
+               Enum.sort(Path.wildcard("#{migration_path}/**/*_migrate_resources*.exs"))
+               |> Enum.reject(&String.contains?(&1, "extensions"))
+
+      contents = File.read!(file2)
+
+      [up_side, down_side] = String.split(contents, "def down", parts: 2)
+
+      up_side_parts = String.split(up_side, "\n", trim: true) |> Enum.map(&String.trim/1)
+
+      # up: drop the constraint before removing the column it covers.
+      assert Enum.find_index(up_side_parts, &(&1 == "drop_if_exists constraint(:posts, :price_must_be_positive)")) <
+               Enum.find_index(up_side_parts, &(&1 == "remove :price"))
+
+      down_side_parts = String.split(down_side, "\n", trim: true) |> Enum.map(&String.trim/1)
+
+      # down (reversed): recreate the column before recreating the
+      # constraint that references it — otherwise this `down` would fail
+      # against real Postgres with "column price does not exist".
+      assert Enum.find_index(down_side_parts, &(&1 == "add :price, :bigint")) <
+               Enum.find_index(down_side_parts, &String.starts_with?(&1, "create constraint(:posts, :price_must_be_positive"))
+    end
+  end
+
+  describe "identity and attribute renamed together" do
+    setup %{snapshot_path: snapshot_path, migration_path: migration_path} do
+      :ok
+
+      defposts do
+        attributes do
+          uuid_primary_key(:id)
+          attribute(:title, :string, public?: true)
+        end
+
+        identities do
+          identity(:uniq_title, [:title])
+        end
+      end
+
+      defdomain([Post])
+
+      AshPostgres.MigrationGenerator.generate(Domain,
+        snapshot_path: snapshot_path,
+        migration_path: migration_path,
+        quiet: true,
+        format: false,
+        auto_name: true
+      )
+    end
+
+    test "the old identity's index is dropped before the rename, so `down` recreates the index after undoing the rename",
+         %{snapshot_path: snapshot_path, migration_path: migration_path} do
+      defposts do
+        attributes do
+          uuid_primary_key(:id)
+          attribute(:title_short, :string, public?: true)
+        end
+
+        identities do
+          identity(:uniq_title, [:title_short])
+        end
+      end
+
+      defdomain([Post])
+
+      send(self(), {:mix_shell_input, :yes?, true})
+
+      AshPostgres.MigrationGenerator.generate(Domain,
+        snapshot_path: snapshot_path,
+        migration_path: migration_path,
+        quiet: true,
+        format: false,
+        auto_name: true
+      )
+
+      assert [_file1, file2] =
+               Enum.sort(Path.wildcard("#{migration_path}/**/*_migrate_resources*.exs"))
+               |> Enum.reject(&String.contains?(&1, "extensions"))
+
+      contents = File.read!(file2)
+
+      [up_side, down_side] = String.split(contents, "def down", parts: 2)
+
+      up_side_parts = String.split(up_side, "\n", trim: true) |> Enum.map(&String.trim/1)
+
+      assert Enum.find_index(up_side_parts, &(&1 == "rename table(:posts), :title, to: :title_short")) <
+               Enum.find_index(up_side_parts, &String.starts_with?(&1, "create unique_index(:posts, [:title_short]"))
+
+      down_side_parts = String.split(down_side, "\n", trim: true) |> Enum.map(&String.trim/1)
+
+      # down (reversed): undo the rename before recreating the old index —
+      # otherwise this `down` would try to index a column name that doesn't
+      # exist yet.
+      assert Enum.find_index(down_side_parts, &(&1 == "rename table(:posts), :title_short, to: :title")) <
+               Enum.find_index(down_side_parts, &String.starts_with?(&1, "create unique_index(:posts, [:title]"))
+    end
+  end
+
   describe "creating follow up migrations with a composite primary key" do
     setup %{snapshot_path: snapshot_path, migration_path: migration_path} do
       :ok
@@ -4879,7 +5029,12 @@ defmodule AshPostgres.MigrationGeneratorTest do
       assert file_content =~
                ~S[add :id, :decimal, null: false, precision: 10, scale: 0, primary_key: true]
 
-      assert file_content =~ ~S[add :category_id, :decimal, null: false, precision: 10, scale: 0]
+      # `category_id` is a new column with its foreign key known from the
+      # start, so it's emitted as a single `add ..., references(...)` (rather
+      # than a separate `add` followed by a later `modify ... references(...)`)
+      # — this still carries the same decimal precision/scale.
+      assert file_content =~
+               ~S[add :category_id, references(:categories, column: :id, name: "products_category_id_fkey", type: :decimal, prefix: "public"), precision: 10, scale: 0, null: false]
     end
   end