@spec bulk_upsert(module(), module(), Enumerable.t(map()), options()) ::
  {:ok, %{upserted: non_neg_integer(), skipped: non_neg_integer()}}

Validate attrs maps (attrs_list) by passing them through an Ecto changeset, then upsert the valid items to the database that corresponds to a given Ecto repo_module (e.g. YourProject.Repo).

attrs_list may be any Enumerable — a plain list, or a lazy Stream for large inputs (see the Streaming section below).

Using a changeset serves two purposes:

1. The changeset can be used to validate and transform the data.
2. Using a changeset allows this function to perform bulk upserts with nested associations.

For validation, each item in the attrs_list is converted to a changeset for a given schema_module. The changeset function is called with a single argument (the attrs map), so the schema module must expose a 1-arity changeset function — e.g. a changeset/2 whose first argument defaults to an empty struct. By default, this function is called :changeset. (See the Options section below for more info.)

Basic example

iex> BulkUpsert.bulk_upsert(
...>   YourProject.Repo,
...>   YourProject.Persons.Person,
...>   _attrs_list = [
...>     %{id: 1, name: "Alice", age: 25, phone_number: "555-1234"},
...>     %{id: 2, name: "Bob", age: 35, phone_number: "555-2345"},
...>   ]
...> )
{:ok, %{upserted: 2, skipped: 0}}

Return value

Returns {:ok, %{upserted: upserted_count, skipped: skipped_count}}, where the counts refer to the top-level attrs: :upserted is the number of items sent to the database, and :skipped is the number of items dropped because their changesets were invalid. (Skipped items are summarized in one :warning log per call, with per-item detail at the :debug level.) A database error raises; by default the entire upsert runs in a single transaction, so every change is rolled back (with :max_concurrency, only the failing chunk is — see below).

Streaming

attrs_list is consumed lazily, in chunks of :chunk_size items: a Stream is never fully materialized, so memory stays bounded for arbitrarily large inputs. Plain lists behave identically (same counts, same single skipped-items summary log). Note that without :max_concurrency, the single transaction — and any locks it takes — stays open for the stream's full duration.

Options

Unknown option names raise an ArgumentError, as does passing a BulkUpsert option (other than :timeout or :placeholders, which insert_all/3 also accepts) inside an :insert_all_opts value.

• :changeset_function_atom - The name of the changeset function to apply for the given schema_module. It is called with one argument: the attrs map. (Default: :changeset)

• :chunk_size - The number of parent attrs items to insert into the database in a single query. Can be increased or decreased as needed to avoid exceeding the Postgres parameter limit for a single query. (Default: 1000)

• :insert_all_function_module - Instead of using the :insert_all function in the given repo_module, you may specify the name of a custom module to use instead. (Default: Inherited from the value specified in the repo_module function argument, e.g. YourProject.Repo)

  • Example: YourProject.OtherRepo

• :insert_all_function_atom - Instead of using your repo module's :insert_all function, you may pass a compatible equivalent that accepts the same arguments. (Default: :insert_all)

  • Example: :insert_all_with_autogenerated_timestamps

• :insert_all_opts - Pass custom opts to the insert_all/3 function. This option consists of a map whose key is the schema or source that may have items being upserted, and the value is the YourProject.Repo.insert_all/3 opts that will be applied when items for that schema are being upserted. By default, a conflicting row has all of its values replaced except the primary key(s) (see the :replace_all_except option). (Default: %{})

  • Example: %{YourProject.Persons.Person => [on_conflict: {:nothing}]}
  • A many_to_many join table is keyed by its source, e.g. %{"persons_topics" => [...]}.
  • :conflict_target defaults to the schema's primary key, so a schema without a primary key must supply its own :conflict_target here (otherwise the upsert fails at the database).

• :max_concurrency - Upsert up to this many chunks of :chunk_size parents concurrently (via Task.async_stream/3), each chunk in its own transaction. By default, all chunks are upserted sequentially inside a single transaction. Setting this option trades the single-transaction guarantee for insert throughput: (Default: nil)

  • A failing chunk still raises, but chunks that already committed stay committed, so a failure partway through leaves the database with partial results.
  • :timeout applies to each chunk's transaction instead of the whole call.
  • Concurrent chunks that share many_to_many child records may upsert the same related (or join table) rows in different orders, which can deadlock in Postgres. Ensure concurrent input does not share child records across chunks, or be prepared to retry on deadlock.
  • In the test environment, the Ecto SQL sandbox requires shared mode (or explicit allowances) so the spawned tasks may use the test's database connection — e.g. use MyApp.DataCase, async: false with a Phoenix-style setup_sandbox/1.

• :placeholders - Set fields from shared values that are sent to the database once instead of once per row, using the :placeholders feature of Ecto's insert_all/3. This option is a map whose key is the schema or source being upserted, and the value is a map of field => value. The fields do not need to appear in the attrs. (Default: %{})

  • Example: %{YourProject.Persons.Person => %{inserted_at: DateTime.utc_now()}}
  • Each placeholder value is injected into the attrs before the changeset is built, so a placeholder field is cast and validated like any other field and may be included in the changeset's validate_required/2.
  • The shared value replaces any per-row value supplied for the field in the attrs.
  • Embedded schemas are stored inline on their parent row and are never upserted as their own source, so placeholder values keyed by an embedded schema module are ignored.

• :recover_changeset_errors - If the given fields in a changeset have errors, then replace them with a custom fallback value. (Default: %{})

  • Example: %{YourProject.Persons.Person => %{phone_number: "INVALID"}}
  • Applies recursively to nested association and embedded changesets, with fallbacks looked up by each changeset's schema (for embeds, the embedded schema module). A parent's association error is cleared once all of that association's child changesets have been recovered.
  • A changeset is only recovered if every one of its error fields has a fallback and every nested changeset is recoverable by the same rule; otherwise the row is skipped.
  • A fallback value is applied without re-running the changeset function, so it must be valid for the schema.
  • Errors on the association and embed fields themselves (e.g. an association whose attrs could not be cast at all) are never recoverable.

• :replace_all_except - If a row already exists, then all fields will be replaced except the primary key, and any fields specified here. (Default: [])

  • Example: [:field, :other_field]

• :timeout - The maximum timeout for the transaction that wraps the entire bulk upsert (all chunks), also applied to each insert_all/3 query. With :max_concurrency, the timeout applies to each chunk's transaction instead. (Default: 15000)

  • Example: 60_000

Examples

Upsert a list of Person attrs using the changeset function YourProject.Persons.Person.upsert_changeset/2 to validate the attrs:

iex> attrs_list = [%{id: 1, name: "Alice", ...}]

iex> BulkUpsert.bulk_upsert(
...>   YourProject.Repo,
...>   YourProject.Persons.Person,
...>   attrs_list,
...>   changeset_function_atom: :upsert_changeset
...> )
{:ok, %{upserted: 1, skipped: 0}}

Upsert a list of attrs, overwriting only the :name field if there is a conflict. Schemas that are not given custom :insert_all_opts keep the default conflict behavior (replace all fields except the primary key):

iex> insert_all_opts = %{
...>   YourProject.Persons.Person => [on_conflict: {:replace, [:name]}]
...> }

iex> BulkUpsert.bulk_upsert(
...>   YourProject.Repo,
...>   YourProject.Persons.Person,
...>   _attrs_list = [%{id: 1, name: "Alicia"}],
...>   insert_all_opts: insert_all_opts
...> )
{:ok, %{upserted: 1, skipped: 0}}

Associations

Nested associations are upserted in the same call as the parent, recursively: a child's own nested associations (at any depth) are upserted the same way as the parent's.

This is an upsert-only operation: rows absent from the attrs are left untouched at every level. Unlike Ecto.Changeset.cast_assoc/3's :on_replace behavior, absent children are never deleted or nilified.

• has_many and has_one: the associated records are upserted into their own table. Each child must include its foreign key in its attrs, since it is upserted directly via insert_all/3.

• many_to_many: the associated records are upserted into their own table, and the join table rows linking each parent to its associations are upserted as well. Duplicate records and links are removed automatically.

• embeds_one and embeds_many: embedded data has no table of its own, so it is stored inline on the parent row as part of the parent upsert.

Known limitations

• Nested belongs_to associations are not upserted. To associate with a belongs_to parent, include its foreign key field in the attrs (e.g. category_id). This applies at every level of nesting.