fix: correlate bulk upsert records by identity on the MERGE path

lib/data_layer.ex

@@ -2449,7 +2449,7 @@ defmodule AshPostgres.DataLayer do
               end)
 
             results =
-              if opts[:upsert?] do
+              if options[:upsert?] do
                 changesets
                 |> Enum.map(fn changeset ->
                   identity =

priv/resource_snapshots/test_repo/bulk_upsert_children/20260703122133.json

@@ -0,0 +1,75 @@
+{
+  "attributes": [
+    {
+      "allow_nil?": false,
+      "default": "fragment(\"gen_random_uuid()\")",
+      "generated?": false,
+      "precision": null,
+      "primary_key?": true,
+      "references": null,
+      "scale": null,
+      "size": null,
+      "source": "id",
+      "type": "uuid"
+    },
+    {
+      "allow_nil?": false,
+      "default": "nil",
+      "generated?": false,
+      "precision": null,
+      "primary_key?": false,
+      "references": null,
+      "scale": null,
+      "size": null,
+      "source": "number",
+      "type": "bigint"
+    },
+    {
+      "allow_nil?": false,
+      "default": "nil",
+      "generated?": false,
+      "precision": null,
+      "primary_key?": false,
+      "references": {
+        "deferrable": false,
+        "destination_attribute": "id",
+        "destination_attribute_default": null,
+        "destination_attribute_generated": null,
+        "index?": false,
+        "match_type": null,
+        "match_with": null,
+        "multitenancy": {
+          "attribute": null,
+          "global": null,
+          "strategy": null
+        },
+        "name": "bulk_upsert_children_parent_id_fkey",
+        "on_delete": null,
+        "on_update": null,
+        "primary_key?": true,
+        "schema": "public",
+        "table": "bulk_upsert_parents"
+      },
+      "scale": null,
+      "size": null,
+      "source": "parent_id",
+      "type": "uuid"
+    }
+  ],
+  "base_filter": null,
+  "check_constraints": [],
+  "create_table_options": null,
+  "custom_indexes": [],
+  "custom_statements": [],
+  "has_create_action": true,
+  "hash": "8C92D71132472AE0C154732115173BB4B0CC41FE03C8847A726450362AC45E2C",
+  "identities": [],
+  "multitenancy": {
+    "attribute": null,
+    "global": null,
+    "strategy": null
+  },
+  "repo": "Elixir.AshPostgres.TestRepo",
+  "schema": null,
+  "table": "bulk_upsert_children"
+}

priv/resource_snapshots/test_repo/bulk_upsert_parents/20260703122134.json

@@ -0,0 +1,71 @@
+{
+  "attributes": [
+    {
+      "allow_nil?": false,
+      "default": "fragment(\"gen_random_uuid()\")",
+      "generated?": false,
+      "precision": null,
+      "primary_key?": true,
+      "references": null,
+      "scale": null,
+      "size": null,
+      "source": "id",
+      "type": "uuid"
+    },
+    {
+      "allow_nil?": false,
+      "default": "nil",
+      "generated?": false,
+      "precision": null,
+      "primary_key?": false,
+      "references": null,
+      "scale": null,
+      "size": null,
+      "source": "number",
+      "type": "bigint"
+    },
+    {
+      "allow_nil?": false,
+      "default": "nil",
+      "generated?": false,
+      "precision": null,
+      "primary_key?": false,
+      "references": null,
+      "scale": null,
+      "size": null,
+      "source": "name",
+      "type": "text"
+    }
+  ],
+  "base_filter": null,
+  "check_constraints": [],
+  "create_table_options": null,
+  "custom_indexes": [],
+  "custom_statements": [],
+  "has_create_action": true,
+  "hash": "8CB14165F3B44262858405FE6ADFB7E16FA1B6C3AC9D08BBFD2266A54405F114",
+  "identities": [
+    {
+      "all_tenants?": false,
+      "base_filter": null,
+      "index_name": "bulk_upsert_parents_unique_number_index",
+      "keys": [
+        {
+          "type": "atom",
+          "value": "number"
+        }
+      ],
+      "name": "unique_number",
+      "nils_distinct?": true,
+      "where": null
+    }
+  ],
+  "multitenancy": {
+    "attribute": null,
+    "global": null,
+    "strategy": null
+  },
+  "repo": "Elixir.AshPostgres.TestRepo",
+  "schema": null,
+  "table": "bulk_upsert_parents"
+}

priv/test_repo/migrations/20260703122132_migrate_resources71.exs

@@ -0,0 +1,51 @@
+defmodule AshPostgres.TestRepo.Migrations.MigrateResources71 do
+  @moduledoc """
+  Updates resources based on their most recent snapshots.
+
+  This file was autogenerated with `mix ash_postgres.generate_migrations`
+  """
+
+  use Ecto.Migration
+
+  def up do
+    create table(:bulk_upsert_parents, primary_key: false) do
+      add(:id, :uuid, null: false, default: fragment("gen_random_uuid()"), primary_key: true)
+      add(:number, :bigint, null: false)
+      add(:name, :text, null: false)
+    end
+
+    create(
+      unique_index(:bulk_upsert_parents, [:number],
+        name: "bulk_upsert_parents_unique_number_index"
+      )
+    )
+
+    create table(:bulk_upsert_children, primary_key: false) do
+      add(:id, :uuid, null: false, default: fragment("gen_random_uuid()"), primary_key: true)
+      add(:number, :bigint, null: false)
+
+      add(
+        :parent_id,
+        references(:bulk_upsert_parents,
+          column: :id,
+          name: "bulk_upsert_children_parent_id_fkey",
+          type: :uuid,
+          prefix: "public"
+        ), null: false)
+    end
+  end
+
+  def down do
+    drop(constraint(:bulk_upsert_children, "bulk_upsert_children_parent_id_fkey"))
+
+    drop(table(:bulk_upsert_children))
+
+    drop_if_exists(
+      unique_index(:bulk_upsert_parents, [:number],
+        name: "bulk_upsert_parents_unique_number_index"
+      )
+    )
+
+    drop(table(:bulk_upsert_parents))
+  end
+end

test/bulk_upsert_managed_relationship_test.exs

@@ -0,0 +1,89 @@
+# SPDX-FileCopyrightText: 2019 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshPostgres.BulkUpsertManagedRelationshipTest do
+  @moduledoc """
+  Regression test for a bulk-upsert managed-relationship association mismatch.
+
+  When `Ash.bulk_create/4` runs a create action with `upsert? true` plus a
+  `manage_relationship/3` change that creates a `has_one` child, each returned
+  parent record must be correlated back to *its own* input changeset before the
+  child is built (the child's foreign key is taken from the parent record).
+
+  On PostgreSQL 17+ the upsert is implemented as a single `MERGE ... RETURNING`
+  statement, whose output rows are **not** guaranteed to be in input order. The
+  data layer previously fell back to zipping the returned rows against the input
+  changesets *by position*, so a reordered `RETURNING` attached children to the
+  wrong parents. Below PG 17 (`INSERT ... ON CONFLICT ... RETURNING`) the order
+  happens to be preserved, which is why the bug only surfaces on the MERGE path.
+
+  Each input `i` builds a parent and a child both tagged with `number: i`, so a
+  correct association always joins a parent to the child carrying the same number.
+  """
+  use AshPostgres.RepoCase, async: false
+
+  alias AshPostgres.Test.BulkUpsertParent
+
+  defp inputs(count) do
+    Enum.map(1..count, fn i ->
+      %{number: i, name: "Parent #{i}", child: %{number: i}}
+    end)
+  end
+
+  defp mismatches do
+    %{rows: rows} =
+      Ecto.Adapters.SQL.query!(
+        AshPostgres.TestRepo,
+        """
+        SELECT p.number, c.number
+        FROM bulk_upsert_parents p
+        JOIN bulk_upsert_children c ON c.parent_id = p.id
+        WHERE p.number <> c.number
+        """,
+        []
+      )
+
+    rows
+  end
+
+  describe "bulk upsert with managed has_one child (MERGE, PostgreSQL 17+)" do
+    @describetag :postgres_17
+
+    test "keeps each child attached to its own parent (fresh insert)" do
+      bulk =
+        Ash.bulk_create(inputs(150), BulkUpsertParent, :upsert_with_child,
+          authorize?: false,
+          return_errors?: true,
+          stop_on_error?: false
+        )
+
+      assert bulk.status == :success
+      assert bulk.errors == []
+      assert mismatches() == []
+    end
+
+    test "keeps each child attached to its own parent (upsert onto existing rows)" do
+      # Pre-populate the parents (without children) in a different physical order
+      # than the later upsert input, maximizing the chance `MERGE ... RETURNING`
+      # returns rows in an order that does not match the bulk input order.
+      parents_only =
+        inputs(150)
+        |> Enum.map(&Map.delete(&1, :child))
+        |> Enum.shuffle()
+
+      Ash.bulk_create!(parents_only, BulkUpsertParent, :upsert_with_child, authorize?: false)
+
+      bulk =
+        Ash.bulk_create(inputs(150), BulkUpsertParent, :upsert_with_child,
+          authorize?: false,
+          return_errors?: true,
+          stop_on_error?: false
+        )
+
+      assert bulk.status == :success
+      assert bulk.errors == []
+      assert mismatches() == []
+    end
+  end
+end

test/support/domain.ex

@@ -77,6 +77,8 @@ defmodule AshPostgres.Test.Domain do
     resource(AshPostgres.Test.ProfileInterest)
     resource(AshPostgres.Test.Label)
     resource(AshPostgres.Test.LabelledContent)
+    resource(AshPostgres.Test.BulkUpsertParent)
+    resource(AshPostgres.Test.BulkUpsertChild)
   end
 
   authorization do

test/support/resources/bulk_upsert_child.ex

@@ -0,0 +1,34 @@
+# SPDX-FileCopyrightText: 2019 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshPostgres.Test.BulkUpsertChild do
+  @moduledoc false
+  use Ash.Resource,
+    domain: AshPostgres.Test.Domain,
+    data_layer: AshPostgres.DataLayer
+
+  postgres do
+    table("bulk_upsert_children")
+    repo(AshPostgres.TestRepo)
+  end
+
+  actions do
+    defaults([:read])
+
+    create :create do
+      accept([:number])
+    end
+  end
+
+  attributes do
+    uuid_primary_key(:id)
+    attribute(:number, :integer, allow_nil?: false, public?: true)
+  end
+
+  relationships do
+    belongs_to :parent, AshPostgres.Test.BulkUpsertParent do
+      allow_nil?(false)
+    end
+  end
+end

test/support/resources/bulk_upsert_parent.ex

@@ -0,0 +1,47 @@
+# SPDX-FileCopyrightText: 2019 ash_postgres contributors <https://github.com/ash-project/ash_postgres/graphs/contributors>
+#
+# SPDX-License-Identifier: MIT
+
+defmodule AshPostgres.Test.BulkUpsertParent do
+  @moduledoc false
+  use Ash.Resource,
+    domain: AshPostgres.Test.Domain,
+    data_layer: AshPostgres.DataLayer
+
+  postgres do
+    table("bulk_upsert_parents")
+    repo(AshPostgres.TestRepo)
+  end
+
+  actions do
+    defaults([:read])
+
+    create :upsert_with_child do
+      upsert?(true)
+      upsert_identity(:unique_number)
+      upsert_fields([:name])
+
+      argument(:child, :map, allow_nil?: true)
+      change(manage_relationship(:child, :child, on_no_match: {:create, :create}))
+
+      accept([:number, :name])
+    end
+  end
+
+  attributes do
+    uuid_primary_key(:id)
+    attribute(:number, :integer, allow_nil?: false, public?: true)
+    attribute(:name, :string, allow_nil?: false, public?: true)
+  end
+
+  identities do
+    identity(:unique_number, [:number])
+  end
+
+  relationships do
+    has_one :child, AshPostgres.Test.BulkUpsertChild do
+      source_attribute(:id)
+      destination_attribute(:parent_id)
+    end
+  end
+end