fix(delta): address critical/warning findings from multi-perspective review

Three bugs surfaced by the /review pass, each with a regression test:

1. RenameTable unusable on follow-up codegen. After a rename, the new
   delta (written to the new table dir) is the only file in that dir;
   reducing it raises ConflictError because state.table (set from
   empty_state to the new name) doesn't match op.old_table.

   Fix: in delta mode, move the old table's delta files into the new
   dir before orphan removal, and make CreateTable.apply_op populate
   state.table from op.table so the full history (CreateTable + ops
   with old name) reduces through the RenameTable cleanly.

2. SetCreateTableOptions was a no-op — squash silently dropped user-set
   storage options like `WITH (fillfactor=70)`.

   Fix: apply_op now writes create_table_options to state, CreateTable
   captures it too, and empty_state/1 exposes the key so squash's
   `initial_operations_for_state` re-emits it.

3. state.empty? was stale after pseudo-op mutations (SetBaseFilter /
   SetCreateTableOptions / SetHasCreateAction). state_empty?/1
   ignored those fields entirely.

   Fix: each pseudo-op's apply_op now sets empty?: false, and
   state_empty?/1 factors in base_filter / create_table_options /
   has_create_action when computing emptiness at load-time.

Also consolidates squash_snapshots.ex's inline initial-state map with
Reducer.empty_state/1 to prevent the three copies from drifting — this
was the underlying cause of the `:create_table_options` KeyError in
the existing squash test after fix #2.

Tests added:
* test/snapshot_delta_test.exs — unit tests for create_table_options
  round-trip through Codec + Reducer + initial_operations_for_state,
  SetCreateTableOptions state propagation, and state_empty?/SetBaseFilter
  consistency.
* test/delta_mode_integration_test.exs — full lifecycle rename +
  follow-up codegen test that reduces the new table dir and asserts
  state is coherent.

Author: barnabasJ

lib/migration_generator/migration_generator.ex

@@ -566,6 +566,14 @@ defmodule AshPostgres.MigrationGenerator do
             create_new_snapshot(snapshots_with_operations, repo_name(repo), opts, tenant?)
           )
           |> then(fn files ->
+            # In delta mode, a rename must move the OLD table's delta files into
+            # the NEW table's directory so the reducer sees the full history
+            # (initial deltas + RenameTable) when it walks the new dir next
+            # codegen. After this move the old dir is empty and the
+            # subsequent `remove_orphan_snapshots` simply deletes the empty
+            # shell. In full-state mode there's only one snapshot file per
+            # table, so the rename does not require a move.
+            move_rename_snapshots(rename_map, repo, opts)
             remove_orphan_snapshots(orphan_snapshots, opts)
             files
           end)
@@ -3661,6 +3669,56 @@ defmodule AshPostgres.MigrationGenerator do
     end
   end
 
+  defp move_rename_snapshots(rename_map, repo, opts) do
+    if opts.dry_run || opts.check || opts.snapshots_only || rename_map == %{} do
+      :ok
+    else
+      if snapshot_format(opts, repo) == :delta do
+        Enum.each(rename_map, fn {{new_table, new_schema},
+                                  %{old_table: old_table, old_schema: old_schema, snapshot: snap}} ->
+          old_snapshot = %{
+            table: old_table,
+            schema: old_schema,
+            repo: repo,
+            multitenancy: snap.multitenancy
+          }
+
+          new_snapshot = %{
+            table: new_table,
+            schema: new_schema,
+            repo: repo,
+            multitenancy: snap.multitenancy
+          }
+
+          old_folder = get_snapshot_folder(old_snapshot, opts)
+          new_folder = get_snapshot_folder(new_snapshot, opts)
+          old_dir = get_snapshot_path(old_snapshot, old_folder)
+          new_dir = get_snapshot_path(new_snapshot, new_folder)
+
+          if File.exists?(old_dir) do
+            File.mkdir_p!(new_dir)
+
+            old_dir
+            |> File.ls!()
+            |> Enum.each(fn file ->
+              src = Path.join(old_dir, file)
+              dst = Path.join(new_dir, file)
+
+              # Skip if destination already exists — should not happen given
+              # rename detection only fires when the new dir is empty, but be
+              # defensive to avoid clobbering.
+              if !File.exists?(dst) do
+                File.rename!(src, dst)
+              end
+            end)
+          end
+        end)
+      end
+
+      :ok
+    end
+  end
+
   defp resolve_renames(_table, adding, [], _opts), do: {adding, [], []}
 
   defp resolve_renames(_table, [], removing, _opts), do: {[], removing, []}

lib/migration_generator/reducer.ex

@@ -102,6 +102,7 @@ defmodule AshPostgres.MigrationGenerator.Reducer do
       base_filter: nil,
       has_create_action: true,
       drop_table_opted_out: false,
+      create_table_options: nil,
       empty?: true,
       multitenancy: %{attribute: nil, strategy: nil, global: nil}
     }
@@ -112,8 +113,14 @@ defmodule AshPostgres.MigrationGenerator.Reducer do
   # =================================================================
 
   defp state_empty?(state) do
+    # Any of these indicate non-emptiness:
+    #   - collections: attributes, identities, indexes, statements, constraints
+    #   - scalar pseudo-op fields: base_filter, create_table_options
+    #   - has_create_action flipped off from the default
     state.attributes == [] and state.identities == [] and state.custom_indexes == [] and
-      state.custom_statements == [] and state.check_constraints == []
+      state.custom_statements == [] and state.check_constraints == [] and
+      is_nil(state.base_filter) and is_nil(state.create_table_options) and
+      state.has_create_action == true
   end
 
   defp snapshot_directory(snapshot, opts) do
@@ -208,12 +215,20 @@ defmodule AshPostgres.MigrationGenerator.Reducer do
   def apply_op(state, %Operation.CreateTable{} = op) do
     # Deltas are persisted in migration-emission order, which places CreateTable
     # AFTER the AddAttribute ops that populate the table. Applying it is
-    # therefore a metadata refresh — it sets the schema, multitenancy, and
-    # repo context but does not require the state to be empty.
+    # therefore a metadata refresh — it sets the table, schema, multitenancy,
+    # create_table_options, and repo context but does not require the state
+    # to be empty.
+    #
+    # Setting state.table from op.table is important for the rename chain:
+    # a subsequent RenameTable op compares state.table to op.old_table, so
+    # state.table must start from the delta's original table name (which may
+    # differ from the current resource's table).
     %{
       state
-      | schema: op.schema,
+      | table: op.table,
+        schema: op.schema,
         multitenancy: op.multitenancy,
+        create_table_options: op.create_table_options,
         empty?: false
     }
     |> maybe_put(:repo, op.repo)
@@ -361,12 +376,14 @@ defmodule AshPostgres.MigrationGenerator.Reducer do
     %{state | check_constraints: Enum.reject(state.check_constraints, &(&1.name == c.name))}
   end
 
-  def apply_op(state, %Operation.SetBaseFilter{new_value: v}), do: %{state | base_filter: v}
+  def apply_op(state, %Operation.SetBaseFilter{new_value: v}),
+    do: %{state | base_filter: v, empty?: false}
 
   def apply_op(state, %Operation.SetHasCreateAction{new_value: v}),
-    do: %{state | has_create_action: v}
+    do: %{state | has_create_action: v, empty?: false}
 
-  def apply_op(state, %Operation.SetCreateTableOptions{new_value: _v}), do: state
+  def apply_op(state, %Operation.SetCreateTableOptions{new_value: v}),
+    do: %{state | create_table_options: v, empty?: false}
 
   def apply_op(state, %Operation.OptOutDropTable{}),
     do: %{state | drop_table_opted_out: true}

lib/mix/tasks/ash_postgres.squash_snapshots.ex

@@ -203,22 +203,17 @@ defmodule Mix.Tasks.AshPostgres.SquashSnapshots do
   end
 
   defp apply_squash({_folder, snapshots, _last_snapshot, into_snapshot, :delta}, _opts) do
-    # Reduce all deltas in timestamp order.
-    initial_state = %{
-      attributes: [],
-      identities: [],
-      schema: nil,
-      custom_indexes: [],
-      custom_statements: [],
-      check_constraints: [],
-      table: nil,
-      repo: nil,
-      base_filter: nil,
-      has_create_action: true,
-      drop_table_opted_out: false,
-      empty?: true,
-      multitenancy: %{attribute: nil, strategy: nil, global: nil}
-    }
+    # Reduce all deltas in timestamp order. Start from the canonical empty
+    # state so every field the reducer might touch is pre-populated — in
+    # particular fields like `:create_table_options` that `apply_op` updates
+    # via map-update syntax (`%{state | key: v}`), which would raise KeyError
+    # on a minimal map.
+    initial_state =
+      AshPostgres.MigrationGenerator.Reducer.empty_state(%{
+        table: nil,
+        schema: nil,
+        repo: nil
+      })
 
     state =
       snapshots

test/delta_mode_integration_test.exs

@@ -459,6 +459,90 @@ defmodule AshPostgres.DeltaModeIntegrationTest do
                op.old_table == "delta_posts_g_old" and op.new_table == "delta_posts_g_new"
              end)
     end
+
+    test "renaming a table, reducing, and generating a follow-up delta succeeds", %{
+      snapshot_path: sp,
+      migration_path: mp
+    } do
+      # Regression test: the previous `test "renaming a table produces a
+      # RenameTable op"` only asserts the op is written into the new dir. It
+      # never reduces that dir afterwards. Today, reducing fails because the
+      # initial deltas stayed in the OLD dir, so the new dir starts from
+      # `empty_state` with table=new but the first op it sees is
+      # RenameTable{old, new} — state.table ≠ op.old_table → ConflictError.
+      #
+      # This test exercises the full lifecycle: rename → codegen with an
+      # additional attribute → assert the reducer sees the combined state.
+      defresource DeltaRenameRegen do
+        postgres do
+          table("delta_rename_regen_old")
+          repo(AshPostgres.TestRepo)
+        end
+
+        actions do
+          defaults([:create, :read, :update, :destroy])
+        end
+
+        attributes do
+          uuid_primary_key(:id)
+          attribute(:title, :string, public?: true)
+        end
+      end
+
+      defdomain([DeltaRenameRegen])
+      run_codegen(Domain, sp, mp)
+
+      defresource DeltaRenameRegen do
+        postgres do
+          table("delta_rename_regen_new")
+          repo(AshPostgres.TestRepo)
+        end
+
+        actions do
+          defaults([:create, :read, :update, :destroy])
+        end
+
+        attributes do
+          uuid_primary_key(:id)
+          attribute(:title, :string, public?: true)
+        end
+      end
+
+      defdomain([DeltaRenameRegen])
+      # "Yes" to "Are you renaming delta_rename_regen_old to delta_rename_regen_new?"
+      send(self(), {:mix_shell_input, :yes?, true})
+      run_codegen(Domain, sp, mp)
+
+      # Now add a column. Codegen must reduce the existing deltas — which
+      # today blows up in Reducer.apply_op(%Operation.RenameTable{}, state)
+      # because state.table != op.old_table.
+      defresource DeltaRenameRegen do
+        postgres do
+          table("delta_rename_regen_new")
+          repo(AshPostgres.TestRepo)
+        end
+
+        actions do
+          defaults([:create, :read, :update, :destroy])
+        end
+
+        attributes do
+          uuid_primary_key(:id)
+          attribute(:title, :string, public?: true)
+          attribute(:body, :string, public?: true)
+        end
+      end
+
+      defdomain([DeltaRenameRegen])
+      run_codegen(Domain, sp, mp)
+
+      state = reduce_dir(sp, "delta_rename_regen_new")
+      sources = Enum.map(state.attributes, & &1.source)
+      assert :id in sources
+      assert :title in sources
+      assert :body in sources
+      assert state.table == "delta_rename_regen_new"
+    end
   end
 
   # ===================================================================