feat: add touch_update_defaults? option to bulk_create (#2590)

Allows callers to skip updating update_default fields (like updated_at)
when a bulk_create upsert results in an update. Supported in ETS and
Mnesia data layers.

- respect explicitly set update_default fields in upsert

When touch_update_defaults? is false, update_default fields that were
explicitly included in upsert_fields or set by the user on the changeset
are now preserved instead of being stripped. Also wires touch_update_defaults?
through Ash.create! for non-bulk upserts via a new upsert/5 data layer callback.

- communicate touch_update_defaults? via changeset context instead of data layer opts

Author: mbaertschi

lib/ash.ex

@@ -365,6 +365,12 @@ defmodule Ash do
                           type: :any,
                           doc:
                             "An expression to check if the record should be updated when there's a conflict."
+                        ],
+                        touch_update_defaults?: [
+                          type: :boolean,
+                          default: true,
+                          doc:
+                            "Whether or not to apply update defaults (like `updated_at` timestamps) on upsert. Only relevant when `upsert?: true` is set. Set to `false` to skip touching update_default fields when an upsert results in an update."
                         ]
                       ]
                       |> Spark.Options.merge(@global_opts, "Global Options")
@@ -679,6 +685,12 @@ defmodule Ash do
                                type: :any,
                                doc:
                                  "An expression to check if the record should be updated when there's a conflict."
+                             ],
+                             touch_update_defaults?: [
+                               type: :boolean,
+                               default: true,
+                               doc:
+                                 "Whether or not to apply update defaults (like `updated_at` timestamps) on upsert. Only relevant when `upsert?: true` is set. Set to `false` to skip touching update_default fields when an upsert results in an update."
                              ]
                            ]
                            |> Spark.Options.merge(

lib/ash/actions/create/bulk.ex

@@ -1298,6 +1298,7 @@ defmodule Ash.Actions.Create.Bulk do
                           nil -> action.return_skipped_upsert?
                           other -> other
                         end,
+                      touch_update_defaults?: Keyword.get(opts, :touch_update_defaults?, true),
                       tenant: Ash.ToTenant.to_tenant(opts[:tenant], resource)
                     }
                   )

lib/ash/actions/create/create.ex

@@ -167,12 +167,15 @@ defmodule Ash.Actions.Create do
     opts =
       Keyword.put(opts, :upsert_identity, upsert_identity)
 
+    touch_update_defaults? = Keyword.get(opts, :touch_update_defaults?, true)
+
     changeset =
       Ash.Changeset.set_context(changeset, %{
         private: %{
           upsert?: true,
           upsert_identity: upsert_identity,
-          upsert_fields: upsert_fields
+          upsert_fields: upsert_fields,
+          touch_update_defaults?: touch_update_defaults?
         }
       })
 

lib/ash/data_layer/data_layer.ex

@@ -203,6 +203,7 @@ defmodule Ash.DataLayer do
             | :replace_all
             | {:replace, list(atom)}
             | {:replace_all_except, list(atom)},
+          touch_update_defaults?: boolean,
           tenant: term()
         }
 

lib/ash/data_layer/ets/ets.ex

@@ -1441,13 +1441,17 @@ defmodule Ash.DataLayer.Ets do
 
   @doc false
   @impl true
-  def upsert(resource, changeset, keys, identity, opts \\ [from_bulk_create?: false]) do
+  def upsert(resource, changeset, keys, identity \\ nil) do
+    do_upsert(resource, changeset, keys, identity)
+  end
+
+  defp do_upsert(resource, changeset, keys, identity, from_bulk_create? \\ false) do
     pkey = Ash.Resource.Info.primary_key(resource)
     keys = keys || pkey
 
     if (is_nil(identity) || !identity.nils_distinct?) &&
          Enum.any?(keys, &is_nil(Ash.Changeset.get_attribute(changeset, &1))) do
-      create(resource, changeset, opts[:from_bulk_create?])
+      create(resource, changeset, from_bulk_create?)
     else
       key_filters =
         Enum.map(keys, fn key ->
@@ -1474,7 +1478,10 @@ defmodule Ash.DataLayer.Ets do
           end
         end)
 
-      to_set = Ash.Changeset.set_on_upsert(changeset, keys)
+      to_set =
+        changeset
+        |> Ash.Changeset.set_on_upsert(keys)
+        |> apply_upsert_update_defaults(resource, changeset)
 
       resource
       |> resource_to_query(changeset.domain)
@@ -1483,7 +1490,7 @@ defmodule Ash.DataLayer.Ets do
       |> run_query(resource)
       |> case do
         {:ok, []} ->
-          create(resource, changeset, opts[:from_bulk_create?])
+          create(resource, changeset, from_bulk_create?)
 
         {:ok, [result]} ->
           with {:ok, conflicting_upsert_values} <- Ash.Changeset.apply_attributes(changeset),
@@ -1503,7 +1510,7 @@ defmodule Ash.DataLayer.Ets do
               resource,
               %{changeset | action_type: :update, filter: nil},
               Map.take(result, pkey),
-              opts[:from_bulk_create?]
+              from_bulk_create?
             )
           else
             {:ok, []} ->
@@ -1519,6 +1526,49 @@ defmodule Ash.DataLayer.Ets do
     end
   end
 
+  defp apply_upsert_update_defaults(to_set, resource, changeset) do
+    touch_update_defaults? =
+      changeset.context[:private][:touch_update_defaults?]
+
+    update_default_attrs =
+      resource
+      |> Ash.Resource.Info.attributes()
+      |> Enum.filter(& &1.update_default)
+
+    if touch_update_defaults? == false || to_set == [] do
+      upsert_fields = changeset.context[:private][:upsert_fields]
+      update_default_names = MapSet.new(update_default_attrs, & &1.name)
+
+      Keyword.reject(to_set, fn {key, _} ->
+        MapSet.member?(update_default_names, key) &&
+          !explicitly_set?(key, upsert_fields, changeset)
+      end)
+    else
+      # Add update_defaults that aren't already in to_set
+      # (set_on_upsert's upsert_fields branch doesn't include them)
+      Enum.reduce(update_default_attrs, to_set, fn attr, acc ->
+        if Keyword.has_key?(acc, attr.name) do
+          acc
+        else
+          value =
+            case attr.update_default do
+              function when is_function(function) -> function.()
+              {m, f, a} when is_atom(m) and is_atom(f) and is_list(a) -> apply(m, f, a)
+              value -> value
+            end
+
+          Keyword.put(acc, attr.name, value)
+        end
+      end)
+    end
+  end
+
+  defp explicitly_set?(key, upsert_fields, _changeset) when is_list(upsert_fields),
+    do: key in upsert_fields
+
+  defp explicitly_set?(key, _, changeset),
+    do: Map.has_key?(changeset.attributes, key) && key not in Map.get(changeset, :defaults, [])
+
   @spec upsert_conflict_check(
           changeset :: Ash.Changeset.t(),
           subject :: record,
@@ -1561,15 +1611,18 @@ defmodule Ash.DataLayer.Ets do
       |> Enum.reduce_while({:ok, []}, fn changeset, {:ok, results} ->
         changeset =
           Ash.Changeset.set_context(changeset, %{
-            private: %{upsert_fields: options[:upsert_fields] || []}
+            private: %{
+              upsert_fields: options[:upsert_fields] || [],
+              touch_update_defaults?: Map.get(options, :touch_update_defaults?, true)
+            }
           })
 
-        case upsert(
+        case do_upsert(
                resource,
                changeset,
                options.upsert_keys,
                options.identity,
-               Map.put(options, :from_bulk_create?, true)
+               true
              ) do
           {:ok, result} ->
             if Ash.Resource.get_metadata(result, :upsert_skipped) do

lib/ash/data_layer/mnesia/mnesia.ex

@@ -353,7 +353,8 @@ defmodule Ash.DataLayer.Mnesia do
           Ash.Changeset.set_context(changeset, %{
             private:
               Map.merge(changeset.context[:private] || %{}, %{
-                upsert_fields: options[:upsert_fields] || []
+                upsert_fields: options[:upsert_fields] || [],
+                touch_update_defaults?: Map.get(options, :touch_update_defaults?, true)
               })
           })
 
@@ -636,7 +637,10 @@ defmodule Ash.DataLayer.Mnesia do
           create(resource, changeset)
 
         {:ok, [result]} ->
-          to_set = Ash.Changeset.set_on_upsert(changeset, keys)
+          to_set =
+            changeset
+            |> Ash.Changeset.set_on_upsert(keys)
+            |> apply_upsert_update_defaults(resource, result, changeset)
 
           changeset =
             changeset
@@ -652,6 +656,40 @@ defmodule Ash.DataLayer.Mnesia do
     end
   end
 
+  # Mnesia's update/2 calls apply_attributes which re-applies update_defaults
+  # via set_defaults/3. To prevent unwanted updates, we preserve existing
+  # values from the record for update_default fields so set_defaults sees
+  # them as already set and skips them.
+  defp apply_upsert_update_defaults(to_set, resource, existing_record, changeset) do
+    touch_update_defaults? =
+      changeset.context[:private][:touch_update_defaults?]
+
+    if touch_update_defaults? == false || to_set == [] do
+      upsert_fields = changeset.context[:private][:upsert_fields]
+
+      update_default_attrs =
+        resource
+        |> Ash.Resource.Info.attributes()
+        |> Enum.filter(& &1.update_default)
+
+      Enum.reduce(update_default_attrs, to_set, fn attr, acc ->
+        if explicitly_set?(attr.name, upsert_fields, changeset) do
+          acc
+        else
+          Keyword.put(acc, attr.name, Map.get(existing_record, attr.name))
+        end
+      end)
+    else
+      to_set
+    end
+  end
+
+  defp explicitly_set?(key, upsert_fields, _changeset) when is_list(upsert_fields),
+    do: key in upsert_fields
+
+  defp explicitly_set?(key, _, changeset),
+    do: Map.has_key?(changeset.attributes, key) && key not in Map.get(changeset, :defaults, [])
+
   @doc false
   @impl true
   def transaction(_, func, _timeout, _reason) do

test/ash/data_layer/ets_test.exs

@@ -45,6 +45,8 @@ defmodule Ash.DataLayer.EtsTest do
       attribute :age, :integer, public?: true
       attribute :title, :string, public?: true
       attribute :roles, {:array, :atom}, public?: true
+      create_timestamp :inserted_at
+      update_timestamp :updated_at, writable?: true, public?: true
     end
   end
 
@@ -138,6 +140,146 @@ defmodule Ash.DataLayer.EtsTest do
              user_table()
   end
 
+  test "upsert with empty upsert_fields does not update update_timestamp" do
+    past = DateTime.add(DateTime.utc_now(), -60, :second)
+
+    %EtsTestUser{id: id} = create_user(%{name: "Mike", updated_at: past})
+
+    updated =
+      create_user(%{name: "Mike Updated", id: id},
+        upsert?: true,
+        upsert_fields: []
+      )
+
+    assert updated.id == id
+    assert updated.name == "Mike"
+    assert DateTime.compare(updated.updated_at, past) == :eq
+  end
+
+  test "upsert does not update update_timestamp when touch_update_defaults? is false" do
+    past = DateTime.add(DateTime.utc_now(), -60, :second)
+
+    %EtsTestUser{id: id} = create_user(%{name: "Mike", updated_at: past})
+
+    updated =
+      create_user(%{name: "Mike Updated", id: id},
+        upsert?: true,
+        touch_update_defaults?: false
+      )
+
+    assert updated.id == id
+    assert updated.name == "Mike Updated"
+    assert DateTime.compare(updated.updated_at, past) == :eq
+  end
+
+  test "upsert preserves explicitly set update_default fields when touch_update_defaults? is false" do
+    past = DateTime.add(DateTime.utc_now(), -120, :second)
+    explicit_time = DateTime.add(DateTime.utc_now(), -30, :second)
+
+    %EtsTestUser{id: id} = create_user(%{name: "Mike", updated_at: past})
+
+    updated =
+      create_user(%{name: "Mike Updated", id: id, updated_at: explicit_time},
+        upsert?: true,
+        touch_update_defaults?: false
+      )
+
+    assert updated.id == id
+    assert updated.name == "Mike Updated"
+    assert DateTime.compare(updated.updated_at, explicit_time) == :eq
+  end
+
+  test "bulk_create with upsert updates update_timestamp" do
+    past = DateTime.add(DateTime.utc_now(), -60, :second)
+
+    %EtsTestUser{id: id} = create_user(%{name: "Mike", updated_at: past})
+
+    assert [{_, %EtsTestUser{updated_at: stored}}] = user_table()
+    assert DateTime.compare(stored, past) == :eq
+
+    result =
+      Ash.bulk_create!(
+        [%{name: "Mike Updated", id: id}],
+        EtsTestUser,
+        :create,
+        upsert?: true,
+        upsert_fields: [:name],
+        return_records?: true
+      )
+
+    assert [%EtsTestUser{id: ^id, name: "Mike Updated", updated_at: new_updated_at}] =
+             result.records
+
+    assert DateTime.after?(new_updated_at, past)
+  end
+
+  test "bulk_create with empty upsert does not update update_timestamp" do
+    past = DateTime.add(DateTime.utc_now(), -60, :second)
+
+    %EtsTestUser{id: id} = create_user(%{name: "Mike", updated_at: past})
+
+    result =
+      Ash.bulk_create!(
+        [%{name: "Mike Updated", id: id}],
+        EtsTestUser,
+        :create,
+        upsert?: true,
+        upsert_fields: [],
+        return_records?: true
+      )
+
+    assert [%EtsTestUser{id: ^id, updated_at: new_updated_at}] = result.records
+    assert DateTime.compare(new_updated_at, past) == :eq
+  end
+
+  test "bulk_create with upsert does not update update_timestamp when touch_update_defaults? is false" do
+    past = DateTime.add(DateTime.utc_now(), -60, :second)
+
+    %EtsTestUser{id: id} = create_user(%{name: "Mike", updated_at: past})
+
+    assert [{_, %EtsTestUser{updated_at: stored}}] = user_table()
+    assert DateTime.compare(stored, past) == :eq
+
+    result =
+      Ash.bulk_create!(
+        [%{name: "Mike Updated", id: id}],
+        EtsTestUser,
+        :create,
+        upsert?: true,
+        upsert_fields: [:name],
+        touch_update_defaults?: false,
+        return_records?: true
+      )
+
+    assert [%EtsTestUser{id: ^id, name: "Mike Updated", updated_at: new_updated_at}] =
+             result.records
+
+    assert DateTime.compare(new_updated_at, past) == :eq
+  end
+
+  test "bulk_create with upsert preserves explicitly set update_default fields when touch_update_defaults? is false" do
+    past = DateTime.add(DateTime.utc_now(), -120, :second)
+    explicit_time = DateTime.add(DateTime.utc_now(), -30, :second)
+
+    %EtsTestUser{id: id} = create_user(%{name: "Mike", updated_at: past})
+
+    result =
+      Ash.bulk_create!(
+        [%{name: "Mike Updated", id: id, updated_at: explicit_time}],
+        EtsTestUser,
+        :create,
+        upsert?: true,
+        upsert_fields: [:name, :updated_at],
+        touch_update_defaults?: false,
+        return_records?: true
+      )
+
+    assert [%EtsTestUser{id: ^id, name: "Mike Updated", updated_at: new_updated_at}] =
+             result.records
+
+    assert DateTime.compare(new_updated_at, explicit_time) == :eq
+  end
+
   test "destroy" do
     mike = create_user(%{name: "Mike"})
     %EtsTestUser{id: joes_id} = joe = create_user(%{name: "Joe"})