return skipped upserts in bulk_create

Author: barnabasJ

lib/data_layer.ex

@@ -2029,6 +2029,66 @@ defmodule AshPostgres.DataLayer do
           repo.insert_all(source, ecto_changesets, opts)
         end)
 
+      result =
+        if options[:return_skipped_upsert?] do
+          identity = options[:identity]
+          [changeset | _] = changesets
+
+          results =
+            result
+            |> elem(1)
+            |> List.wrap()
+            |> Enum.reduce(%{}, fn r, acc ->
+              Map.put(acc, Map.take(r, identity.keys), r)
+            end)
+
+          ash_query =
+            resource
+            |> Ash.Query.do_filter(
+              or:
+                changesets
+                |> Enum.filter(fn changeset ->
+                  not Map.has_key?(results, Map.take(changeset.attributes, identity.keys))
+                end)
+                |> Enum.map(fn changeset ->
+                  changeset.attributes
+                  |> Map.take(identity.keys)
+                  |> Keyword.new()
+                end)
+            )
+            |> then(fn
+              query when is_nil(identity) or is_nil(identity.where) -> query
+              query -> Ash.Query.do_filter(query, identity.where)
+            end)
+            |> Ash.Query.set_tenant(changeset.tenant)
+
+          skipped_upserts =
+            with {:ok, ecto_query} <- Ash.Query.data_layer_query(ash_query),
+                 {:ok, results} <- run_query(ecto_query, resource) do
+              results
+              |> Enum.map(fn result ->
+                Ash.Resource.put_metadata(result, :upsert_skipped, true)
+              end)
+              |> Enum.reduce(%{}, fn r, acc ->
+                Map.put(acc, Map.take(r, identity.keys), r)
+              end)
+            end
+
+          results =
+            changesets
+            |> Enum.map(fn changeset ->
+              identity =
+                changeset.attributes
+                |> Map.take(identity.keys)
+
+              Map.get(results, identity, Map.get(skipped_upserts, identity))
+            end)
+
+          {length(results), results}
+        else
+          result
+        end
+
       case result do
         {_, nil} ->
           :ok
@@ -2039,6 +2099,7 @@ defmodule AshPostgres.DataLayer do
 
             {:ok, results}
           else
+            # TODO: what if there are less results than changesets because of upsert conditions?
             {:ok,
              Stream.zip_with(results, changesets, fn result, changeset ->
                if !opts[:upsert?] do

mix.exs

@@ -169,7 +169,7 @@ defmodule AshPostgres.MixProject do
       {:ash, ash_version("~> 3.5 and >= 3.5.35")},
       {:spark, "~> 2.3 and >= 2.3.4"},
       {:ash_sql, ash_sql_version("~> 0.2 and >= 0.2.90")},
-      {:igniter, "~> 0.6 and >= 0.6.14", optional: true},
+      {:igniter, "~> 0.6 and >= 0.6.29", optional: true},
       {:ecto_sql, "~> 3.13"},
       {:ecto, "~> 3.13"},
       {:jason, "~> 1.0"},
@@ -186,7 +186,7 @@ defmodule AshPostgres.MixProject do
       {:credo, ">= 0.0.0", only: [:dev, :test], runtime: false},
       {:mix_audit, ">= 0.0.0", only: [:dev, :test], runtime: false},
       {:dialyxir, ">= 0.0.0", only: [:dev, :test], runtime: false},
-      {:sobelow, ">= 0.0.0", only: [:dev, :test], runtime: false}
+      {:mix_test_watch, "~> 1.0", only: [:dev, :test]}
     ]
   end
 

mix.lock

@@ -33,6 +33,7 @@
   "mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"},
   "mint": {:hex, :mint, "1.7.1", "113fdb2b2f3b59e47c7955971854641c61f378549d73e829e1768de90fc1abf1", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1 or ~> 0.2.0 or ~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "fceba0a4d0f24301ddee3024ae116df1c3f4bb7a563a731f45fdfeb9d39a231b"},
   "mix_audit": {:hex, :mix_audit, "2.1.5", "c0f77cee6b4ef9d97e37772359a187a166c7a1e0e08b50edf5bf6959dfe5a016", [:make, :mix], [{:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}, {:yaml_elixir, "~> 2.11", [hex: :yaml_elixir, repo: "hexpm", optional: false]}], "hexpm", "87f9298e21da32f697af535475860dc1d3617a010e0b418d2ec6142bc8b42d69"},
+  "mix_test_watch": {:hex, :mix_test_watch, "1.3.0", "2ffc9f72b0d1f4ecf0ce97b044e0e3c607c3b4dc21d6228365e8bc7c2856dc77", [:mix], [{:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}], "hexpm", "f9e5edca976857ffac78632e635750d158df14ee2d6185a15013844af7570ffe"},
   "nimble_options": {:hex, :nimble_options, "1.1.1", "e3a492d54d85fc3fd7c5baf411d9d2852922f66e69476317787a7b2bb000a61b", [:mix], [], "hexpm", "821b2470ca9442c4b6984882fe9bb0389371b8ddec4d45a9504f00a66f650b44"},
   "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"},
   "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"},

test/bulk_create_test.exs

@@ -171,6 +171,71 @@ defmodule AshPostgres.BulkCreateTest do
                end)
     end
 
+    @tag :focus
+    test "bulk upsert returns skipped records with return_skipped_upsert?" do
+      assert [
+               {:ok, %{title: "fredfoo", uniq_if_contains_foo: "1foo", price: 10}},
+               {:ok, %{title: "georgefoo", uniq_if_contains_foo: "2foo", price: 20}},
+               {:ok, %{title: "herbert", uniq_if_contains_foo: "3", price: 30}}
+             ] =
+               Ash.bulk_create!(
+                 [
+                   %{title: "fredfoo", uniq_if_contains_foo: "1foo", price: 10},
+                   %{title: "georgefoo", uniq_if_contains_foo: "2foo", price: 20},
+                   %{title: "herbert", uniq_if_contains_foo: "3", price: 30}
+                 ],
+                 Post,
+                 :create,
+                 return_stream?: true,
+                 return_records?: true
+               )
+               |> Enum.sort_by(fn {:ok, result} -> result.title end)
+
+      results =
+        Ash.bulk_create!(
+          [
+            %{title: "fredfoo", uniq_if_contains_foo: "1foo", price: 10},
+            %{title: "georgefoo", uniq_if_contains_foo: "2foo", price: 20_000},
+            %{title: "herbert", uniq_if_contains_foo: "3", price: 30}
+          ],
+          Post,
+          :upsert_with_no_filter,
+          return_stream?: true,
+          upsert_condition: expr(price != upsert_conflict(:price)),
+          return_errors?: true,
+          return_records?: true,
+          return_skipped_upsert?: true
+        )
+        |> Enum.sort_by(fn
+          {:ok, result} ->
+            result.title
+
+          _ ->
+            nil
+        end)
+
+      assert [
+               {:ok, skipped},
+               {:ok, updated},
+               {:ok, no_conflict}
+             ] = results
+
+      # "fredfoo" was skipped because price matches (10 == 10)
+      assert skipped.title == "fredfoo"
+      assert skipped.price == 10
+      assert Ash.Resource.get_metadata(skipped, :upsert_skipped) == true
+
+      # "georgefoo" was updated because price differs (20 -> 20_000)
+      assert updated.title == "georgefoo"
+      assert updated.price == 20_000
+      refute Ash.Resource.get_metadata(updated, :upsert_skipped)
+
+      # "herbert" had no conflict (doesn't match identity)
+      assert no_conflict.title == "herbert"
+      assert no_conflict.price == 30
+      refute Ash.Resource.get_metadata(no_conflict, :upsert_skipped)
+    end
+
     # confirmed that this doesn't work because it can't. An upsert must map to a potentially successful insert.
     # leaving this test here for posterity
     # test "bulk creates can upsert with id" do