defmodule Bulkinup do @moduledoc """ Bulk inserts and upserts for nested Ecto schemas. Ecto's `c:Ecto.Repo.insert_all/3` writes one flat table at a time and skips changeset validation. Bulkinup fills that gap: it validates attrs maps through your schemas' changeset functions, then writes a parent **and its nested associations across multiple tables** in one call, wrapped in a single transaction by default. Two sibling verbs share the same engine, options, and semantics: - `upsert/4` — bulk upsert: existing rows are updated on conflict. - `insert/4` — pure bulk insert: writing a row that already exists raises. `use Bulkinup` (see `__using__/1`) injects repo-scoped `bulk_insert/3` and `bulk_upsert/3` into your repo module, with app-wide defaults declared once at `use` time. For a walkthrough, see the guides: [Getting Started](getting_started.html), [Nested Associations](nested_associations.html), [Recipes](recipes.html), and [Migrating from bulk_upsert](migrating_from_bulk_upsert.html). """ require Logger @default_timeout 15_000 # Cap the number of skipped-item IDs included in the summary warning's metadata @skipped_item_ids_log_limit 50 @valid_options [ :changeset_function_atom, :chunk_size, :insert_all_function_atom, :insert_all_function_module, :insert_all_opts, :max_concurrency, :placeholders, :recover_changeset_errors, :replace_all_except, :timeout ] # `:timeout` and `:placeholders` are also valid `insert_all/3` options, so they may appear # inside `:insert_all_opts` values as well @options_misplaced_inside_insert_all_opts @valid_options -- [:timeout, :placeholders] # Options that only make sense for an upsert; `insert/4` raises when given one @upsert_only_options [:replace_all_except] # The verbs, as they appear in the `use Bulkinup` per-verb namespaces @verbs [:insert, :upsert] @typedoc """ Options accepted by `insert/4` and `upsert/4`. See `upsert/4`'s documentation for details. `:replace_all_except` is upsert-only: `insert/4` raises an `ArgumentError` when given it. Map keys typed `module() | Ecto.Schema.source()` accept a schema module or, for `many_to_many` join tables, the source as a string (e.g. `"persons_topics"`). """ @type options :: [ changeset_function_atom: atom(), chunk_size: pos_integer(), insert_all_function_module: module(), insert_all_function_atom: atom(), insert_all_opts: %{optional(module() | Ecto.Schema.source()) => Keyword.t()}, max_concurrency: pos_integer(), placeholders: %{ optional(module() | Ecto.Schema.source()) => %{optional(atom()) => term()} }, recover_changeset_errors: %{optional(module()) => %{optional(atom()) => term()}}, replace_all_except: [atom()], timeout: timeout() ] @doc """ 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> Bulkinup.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 Bulkinup option (other than `:timeout` or `:placeholders`, which `insert_all/3` also accepts) inside an `:insert_all_opts` value. > #### Warning {: .warning} > > The `:changeset_function_atom`, `:insert_all_function_module`, and `:insert_all_function_atom` > options are invoked via `apply/3`. Never build these option values from untrusted (e.g. > user-supplied) input. - `: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: `#{@default_timeout}`) - 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> Bulkinup.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> Bulkinup.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. """ @spec upsert(module(), module(), Enumerable.t(map()), options()) :: {:ok, %{upserted: non_neg_integer(), skipped: non_neg_integer()}} def upsert(repo_module, schema_module, attrs_list, opts \\ []) do bulk_write(:upsert, repo_module, schema_module, attrs_list, opts) end @doc """ Like `upsert/4`, but a pure bulk insert: rows are only ever created, never updated. No `on_conflict` or `conflict_target` defaults are applied at any level — the parent schema, nested associations, and `many_to_many` join tables all use Ecto's default conflict behavior, so inserting a row (or join table link) that already exists raises (a `Postgrex.Error` unique violation on Postgres). By default the entire insert runs in a single transaction, so every change is rolled back when a duplicate raises. To tolerate children or join rows shared with data that is already in the database (e.g. `many_to_many` records that several parents reference), override the conflict behavior for just those sources via `:insert_all_opts`: iex> Bulkinup.insert( ...> YourProject.Repo, ...> YourProject.Blog.Post, ...> attrs_list, ...> insert_all_opts: %{ ...> YourProject.Blog.Tag => [on_conflict: :nothing], ...> "posts_tags" => [on_conflict: :nothing] ...> } ...> ) {:ok, %{inserted: 2, skipped: 0}} Returns `{:ok, %{inserted: inserted_count, skipped: skipped_count}}`, with the same meaning as `upsert/4`'s counts. Everything else — changeset validation, `:recover_changeset_errors`, `:placeholders`, chunking, streaming input, `:max_concurrency`, `:timeout`, and the skipped-items summary logging — behaves exactly as documented for `upsert/4`. The upsert-only option `:replace_all_except` raises an `ArgumentError`. """ @spec insert(module(), module(), Enumerable.t(map()), options()) :: {:ok, %{inserted: non_neg_integer(), skipped: non_neg_integer()}} def insert(repo_module, schema_module, attrs_list, opts \\ []) do bulk_write(:insert, repo_module, schema_module, attrs_list, opts) end @doc """ Injects `bulk_insert/3` and `bulk_upsert/3` into the calling repo module — `insert/4` and `upsert/4` scoped to that repo, with shared defaults declared once at `use` time: defmodule YourProject.Repo do use Ecto.Repo, otp_app: :your_project, adapter: Ecto.Adapters.Postgres use Bulkinup, timeout: fetch_timeout!(), insert: [chunk_size: 500], upsert: [replace_all_except: [:inserted_at]] end YourProject.Repo.bulk_upsert(Person, attrs_list) {:ok, %{upserted: 2, skipped: 0}} ## Defaults and precedence - A flat key (e.g. `timeout:` above) is a shared default, applied to each verb it is valid for. The upsert-only `:replace_all_except` is applied to `bulk_upsert/3` only — though passing it per-call to `bulk_insert/3` still raises, as it does for `insert/4`. - The `insert:` and `upsert:` keys hold per-verb defaults. - Precedence: per-call opts override the verb namespace, which overrides flat keys. Option names are validated at compile time: a typo, or a key that is valid for no verb it would apply to, is a compile error. Option *values* are injected unevaluated and evaluated per call at runtime, so dynamic defaults (like `fetch_timeout!()` above, resolved in the repo module) work. """ defmacro __using__(use_opts) do {flat_opts, verb_opts} = validate_use_opts!(use_opts) insert_defaults = build_verb_defaults(:insert, flat_opts, verb_opts) upsert_defaults = build_verb_defaults(:upsert, flat_opts, verb_opts) quote do @doc "`Bulkinup.insert/4` scoped to this repo, with the `use Bulkinup` defaults applied." def bulk_insert(schema_module, attrs_list, opts \\ []) do defaults = unquote(insert_defaults) Bulkinup.insert(__MODULE__, schema_module, attrs_list, Keyword.merge(defaults, opts)) end @doc "`Bulkinup.upsert/4` scoped to this repo, with the `use Bulkinup` defaults applied." def bulk_upsert(schema_module, attrs_list, opts \\ []) do defaults = unquote(upsert_defaults) Bulkinup.upsert(__MODULE__, schema_module, attrs_list, Keyword.merge(defaults, opts)) end end end # Validate the `use Bulkinup` option names at compile time. Values are AST and are not # evaluated here — they are injected into the generated functions and evaluated per call. defp validate_use_opts!(use_opts) do if not Keyword.keyword?(use_opts) do raise ArgumentError, """ `use Bulkinup` expects a literal keyword list of options, \ got: #{Macro.to_string(use_opts)}\ """ end {verb_opts, flat_opts} = Keyword.split(use_opts, @verbs) unknown_options = flat_opts |> Keyword.keys() |> Enum.uniq() |> Kernel.--(@valid_options) if unknown_options != [] do raise ArgumentError, """ unknown option(s) #{inspect(unknown_options)} given to `use Bulkinup`. Valid options: #{inspect(@valid_options)}, plus the #{inspect(@verbs)} per-verb \ namespaces\ """ end Enum.each(verb_opts, fn {verb, verb_defaults} -> if not Keyword.keyword?(verb_defaults) do raise ArgumentError, """ the `use Bulkinup` option `#{inspect(verb)}:` expects a literal keyword list of \ defaults for that verb, got: #{Macro.to_string(verb_defaults)}\ """ end valid_verb_options = valid_options_for_verb(verb) invalid_options = verb_defaults |> Keyword.keys() |> Enum.uniq() |> Kernel.--(valid_verb_options) if invalid_options != [] do raise ArgumentError, """ invalid option(s) #{inspect(invalid_options)} in the `#{inspect(verb)}:` namespace \ given to `use Bulkinup`. Valid options for `#{inspect(verb)}:`: #{inspect(valid_verb_options)}\ """ end end) {flat_opts, verb_opts} end defp valid_options_for_verb(:insert), do: @valid_options -- @upsert_only_options defp valid_options_for_verb(:upsert), do: @valid_options # Merge a verb's defaults at compile time (the values stay unevaluated AST): flat keys the # verb accepts, overridden by the verb's namespace defp build_verb_defaults(verb, flat_opts, verb_opts) do flat_opts |> Keyword.take(valid_options_for_verb(verb)) |> Keyword.merge(Keyword.get(verb_opts, verb, [])) end # The shared write engine: validate, chunk, write, and summarize. Verb-specific behavior # (conflict defaults and the verb's count key) dispatches on `config.verb`. defp bulk_write(verb, repo_module, schema_module, attrs_list, opts) do validate_opts!(verb, opts) # Parse all options once; `config` is threaded through every helper below config = %{ verb: verb, changeset_function_atom: Keyword.get(opts, :changeset_function_atom, :changeset), chunk_size: Keyword.get(opts, :chunk_size, 1000), max_concurrency: Keyword.get(opts, :max_concurrency), recover_changeset_errors: Keyword.get(opts, :recover_changeset_errors, %{}), insert_all_function_module: Keyword.get(opts, :insert_all_function_module, repo_module), insert_all_function_atom: Keyword.get(opts, :insert_all_function_atom, :insert_all), insert_all_opts: Keyword.get(opts, :insert_all_opts, %{}), replace_all_except: Keyword.get(opts, :replace_all_except, []), placeholders: Keyword.get(opts, :placeholders, %{}), timeout: Keyword.get(opts, :timeout, @default_timeout) } attrs_list = if config.placeholders == %{} do attrs_list else # Inject each placeholder value 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 # `validate_required/2`). The value is swapped for a `{:placeholder, field}` tuple after # validation, in `insert_all_entries/4` Stream.map(attrs_list, &inject_placeholder_values(&1, schema_module, config.placeholders)) end # Build changesets lazily and chunk large payloads to stay within Postgres bulk limits. # Laziness lets any `Enumerable` (including a `Stream`) be validated and written # chunk-by-chunk without materializing the whole input changeset_chunks = attrs_list # Convert to changesets so the data can be validated before writing |> Stream.map(fn attrs -> apply(schema_module, config.changeset_function_atom, [attrs]) end) |> Stream.chunk_every(config.chunk_size) totals = changeset_chunks |> write_chunks(repo_module, schema_module, config) |> aggregate_chunk_results() if totals.skipped > 0, do: log_skipped_changesets_summary(schema_module, totals, verb) {:ok, %{count_key(verb) => totals.written, skipped: totals.skipped}} end # The key the verb's written count is returned under defp count_key(:insert), do: :inserted defp count_key(:upsert), do: :upserted # The verb as it appears in log prose defp verb_past_tense(:insert), do: "inserted" defp verb_past_tense(:upsert), do: "upserted" # Write every chunk sequentially, wrapped in a single transaction so that any failure rolls # back all changes made to every chunk of parents and all of their associations defp write_chunks( changeset_chunks, repo_module, schema_module, %{max_concurrency: nil} = config ) do {:ok, chunk_results} = repo_module.transaction( fn -> Enum.map(changeset_chunks, &write_chunk(schema_module, &1, config)) end, timeout: config.timeout ) chunk_results end # Write chunks concurrently, each in its own transaction. A failing chunk raises in the # caller, but chunks that already committed stay committed (see the `:max_concurrency` docs) defp write_chunks(changeset_chunks, repo_module, schema_module, config) do changeset_chunks |> Task.async_stream( fn changesets -> # An exception is returned instead of raised, then reraised in the caller below, so a # failing chunk propagates the original error (e.g. a `Postgrex.Error`) just like the # sequential mode, rather than exiting the caller try do {:ok, chunk_result} = repo_module.transaction( fn -> write_chunk(schema_module, changesets, config) end, timeout: config.timeout ) {:ok, chunk_result} rescue exception -> {:raised, exception, __STACKTRACE__} end end, max_concurrency: config.max_concurrency, timeout: :infinity ) |> Enum.map(fn {:ok, {:ok, chunk_result}} -> chunk_result {:ok, {:raised, exception, stacktrace}} -> reraise(exception, stacktrace) end) end # Validate, recover, and write one chunk of parent changesets, returning the chunk's counts # and the primary keys of its skipped items (capped, for the end-of-call summary log) defp write_chunk(schema_module, changesets, config) do {valid_changesets, invalid_changesets} = changesets |> recover_changesets_with_recoverable_errors(config.recover_changeset_errors) |> Enum.split_with(& &1.valid?) Enum.each(invalid_changesets, &log_on_changeset_error(schema_module, &1, config.verb)) if valid_changesets != [], do: do_bulk_write(schema_module, valid_changesets, config) %{ written: length(valid_changesets), skipped: length(invalid_changesets), skipped_item_ids: invalid_changesets |> Enum.take(@skipped_item_ids_log_limit) |> Enum.map(&changeset_primary_key(schema_module, &1)) } end # Sum the per-chunk counts, keeping the skipped-item IDs capped so an arbitrarily long input # cannot accumulate unbounded log metadata defp aggregate_chunk_results(chunk_results) do initial_totals = %{written: 0, skipped: 0, skipped_item_ids: []} Enum.reduce(chunk_results, initial_totals, fn chunk_result, totals -> remaining_id_slots = @skipped_item_ids_log_limit - length(totals.skipped_item_ids) %{ written: totals.written + chunk_result.written, skipped: totals.skipped + chunk_result.skipped, skipped_item_ids: totals.skipped_item_ids ++ Enum.take(chunk_result.skipped_item_ids, remaining_id_slots) } end) end defp changeset_primary_key(schema_module, changeset) do schema_module.__schema__(:primary_key) |> Map.new(fn primary_key_field -> {primary_key_field, changeset.changes[primary_key_field]} end) end # Raise on unknown option names, on upsert-only options given to `insert/4`, and on # Bulkinup-level options nested inside `:insert_all_opts` values. The keys of each # `:insert_all_opts` value are otherwise not checked, since the set of valid `insert_all/3` # options belongs to Ecto. defp validate_opts!(verb, opts) do unknown_options = opts |> Keyword.keys() |> Enum.uniq() |> Kernel.--(@valid_options) if unknown_options != [] do raise ArgumentError, """ unknown option(s) #{inspect(unknown_options)}. Valid options: #{inspect(@valid_options)}\ """ end upsert_only_options = if verb == :insert, do: opts |> Keyword.keys() |> Enum.uniq() |> Enum.filter(&(&1 in @upsert_only_options)), else: [] if upsert_only_options != [] do raise ArgumentError, """ the option(s) #{inspect(upsert_only_options)} only apply to an upsert, and are not \ supported by `insert/4`. Use `upsert/4` instead, or drop the option(s).\ """ end chunk_size = Keyword.get(opts, :chunk_size, 1000) if not (is_integer(chunk_size) and chunk_size > 0) do raise ArgumentError, """ the `:chunk_size` option must be a positive integer, got: #{inspect(chunk_size)}\ """ end max_concurrency = Keyword.get(opts, :max_concurrency) if not (is_nil(max_concurrency) or (is_integer(max_concurrency) and max_concurrency > 0)) do raise ArgumentError, """ the `:max_concurrency` option must be a positive integer, got: #{inspect(max_concurrency)}\ """ end insert_all_opts = Keyword.get(opts, :insert_all_opts, %{}) if not is_map(insert_all_opts) do raise ArgumentError, """ the `:insert_all_opts` option must be a map of `schema_or_source => insert_all_opts`, \ got: #{inspect(insert_all_opts)}\ """ end Enum.each(insert_all_opts, fn {schema_or_source, schema_insert_all_opts} -> misplaced_options = if Keyword.keyword?(schema_insert_all_opts), do: schema_insert_all_opts |> Keyword.keys() |> Enum.uniq() |> Enum.filter(&(&1 in @options_misplaced_inside_insert_all_opts)), else: [] if misplaced_options != [] do raise ArgumentError, """ the option(s) #{inspect(misplaced_options)} given for #{inspect(schema_or_source)} are \ Bulkinup options, and have no effect inside `:insert_all_opts`. Pass them at the top \ level of `opts` instead.\ """ end end) end # Inject the placeholder values configured for a schema into an attrs map, recursing into the # attrs of every association that will be upserted. The shared placeholder value replaces any # per-row value for the field. defp inject_placeholder_values(attrs, schema_module, placeholders) do attrs = placeholders |> Map.get(schema_module, %{}) |> Enum.reduce(attrs, fn {field, value}, acc -> put_placeholder_attr(acc, field, value) end) # Recurse into the associations that are upserted (`has_many`, `has_one`, `many_to_many`) (get_schema_associations(schema_module, :has) ++ get_schema_associations(schema_module, :many_to_many)) |> Enum.reduce(attrs, fn association, acc -> related_schema_module = schema_module.__schema__(:association, association).related update_association_attrs( acc, association, &inject_placeholder_values(&1, related_schema_module, placeholders) ) end) end # Attrs maps may use string or atom keys, and Ecto's `cast/4` raises on maps that mix both, so # the injected key must match the keys already present. defp put_placeholder_attr(attrs, field, value) do if Enum.any?(Map.keys(attrs), &is_binary/1), do: Map.put(attrs, Atom.to_string(field), value), else: Map.put(attrs, field, value) end # Apply `fun` to each attrs map under an association key, if the association is present. # Values that are not attrs maps (or lists of them) are left untouched, so the changeset cast # reports them as errors in the usual way. defp update_association_attrs(attrs, association, fun) do key = cond do Map.has_key?(attrs, association) -> association Map.has_key?(attrs, Atom.to_string(association)) -> Atom.to_string(association) true -> nil end if is_nil(key) do attrs else Map.update!(attrs, key, fn children when is_list(children) -> Enum.map(children, fn child when is_map(child) -> fun.(child) not_a_map -> not_a_map end) child when is_map(child) -> fun.(child) not_attrs -> not_attrs end) end end defp attrs_from_changeset(changeset) do struct = Ecto.Changeset.apply_action!(changeset, :bulk_write) struct |> Map.from_struct() |> Map.reject(fn {k, _v} -> k not in Map.keys(changeset.changes) end) end # Upsert a list of entries into a schema or source, applying any configured placeholders and # chunking large payloads to stay within Postgres bulk limits. defp insert_all_entries(entries, schema_or_source, insert_all_opts, context) do placeholder_values = context.placeholders[schema_or_source] || %{} # Placeholder fields are set here, after changeset validation, because a `{:placeholder, key}` # tuple cannot pass through a changeset. Each placeholder value is sent to Postgres once. {entries, insert_all_opts} = if placeholder_values == %{} do {entries, insert_all_opts} else placeholder_attrs = Map.new(placeholder_values, fn {field, _value} -> {field, {:placeholder, field}} end) { Enum.map(entries, &Map.merge(&1, placeholder_attrs)), Keyword.put(insert_all_opts, :placeholders, placeholder_values) } end entries |> Enum.chunk_every(context.chunk_size) |> Enum.each(fn entries_chunk -> apply(context.insert_all_function_module, context.insert_all_function_atom, [ schema_or_source, entries_chunk, insert_all_opts ]) end) end # Write one schema's rows and recurse into its associations, applying the verb's default # conflict behavior (see `default_schema_insert_all_opts/3`) at every level defp do_bulk_write(schema_module, changesets, config) do %{insert_all_opts: insert_all_opts} = config # Perform the bulk write for all parent attrs attrs_list = changesets # Drop all assoc data from the changeset (assocs are handled separately in a later step) |> Enum.map(&drop_association_changes(&1, schema_module)) |> Enum.map(&attrs_from_changeset/1) # Build `insert_all` opts for the parent schema parent_insert_all_opts = Keyword.merge( default_schema_insert_all_opts(config.verb, schema_module, config), insert_all_opts[schema_module] || [] ) insert_all_entries(attrs_list, schema_module, parent_insert_all_opts, config) # Perform the bulk write for all `has_many` and `has_one` associations for association <- get_schema_associations(schema_module, :has) do association_schema_module = schema_module.__changeset__()[association] |> elem(1) |> Map.fetch!(:related) association_changesets = changesets |> Enum.map(& &1.changes) |> Enum.map(&Map.get(&1, association)) # `has_many` changes are a list of changesets; `has_one` changes are a single # changeset; an absent association is `nil`. `List.wrap/1` normalizes each into a # (possibly empty) list, which `flat_map` concatenates. |> Enum.flat_map(&List.wrap/1) # Recurse so each child's own nested associations are written as well do_bulk_write(association_schema_module, association_changesets, config) end # Perform the bulk write for all `many_to_many` associations for association <- get_schema_associations(schema_module, :many_to_many) do %Ecto.Association.ManyToMany{ related: related_schema_module, join_through: join_through, join_keys: [{owner_join_key, owner_key}, {related_join_key, related_key}] } = schema_module.__changeset__()[association] |> elem(1) # Pair each parent's primary key with each of its related changesets, so the related # records and the join rows can both be derived from the same data. parent_related_pairs = Enum.flat_map(changesets, fn parent_changeset -> parent_changeset.changes |> Map.get(association) |> List.wrap() |> Enum.map(fn related_changeset -> {parent_changeset, related_changeset} end) end) # Write the related records into their own table. The same record may be referenced by # multiple parents, so duplicates are removed to avoid conflicting twice in one query. related_changesets = parent_related_pairs |> Enum.map(fn {_parent_changeset, related_changeset} -> related_changeset end) |> Enum.uniq_by(fn related_changeset -> related_schema_module.__schema__(:primary_key) |> Enum.map(&Ecto.Changeset.get_field(related_changeset, &1)) end) # Recurse so each related record's own nested associations are written as well do_bulk_write(related_schema_module, related_changesets, config) # Write the join table rows that link each parent to its related records. The same link # may be listed more than once, so duplicate rows are removed for the same reason. join_attrs_list = parent_related_pairs |> Enum.map(fn {parent_changeset, related_changeset} -> %{ owner_join_key => Ecto.Changeset.fetch_field!(parent_changeset, owner_key), related_join_key => Ecto.Changeset.fetch_field!(related_changeset, related_key) } end) |> Enum.uniq() join_insert_all_opts = Keyword.merge( default_join_insert_all_opts(config.verb, [owner_join_key, related_join_key], config), insert_all_opts[join_through] || [] ) insert_all_entries( join_attrs_list, join_through, join_insert_all_opts, config ) end end # Default `insert_all` opts for a schema's own rows, per verb. An insert sets no conflict # options at all, so a duplicate row raises (Ecto's default behavior); an upsert conflicts on # the primary key and replaces every other field (minus `:replace_all_except`). defp default_schema_insert_all_opts(:insert, _schema_module, config) do [timeout: config.timeout] end defp default_schema_insert_all_opts(:upsert, schema_module, config) do [ conflict_target: schema_module.__schema__(:primary_key), on_conflict: {:replace_all_except, schema_module.__schema__(:primary_key) ++ config.replace_all_except}, timeout: config.timeout ] end # Default `insert_all` opts for a `many_to_many` join table's rows, per verb. An insert sets # no conflict options here either, so a duplicate link raises just like a duplicate row; an # upsert ignores links that already exist. defp default_join_insert_all_opts(:insert, _join_keys, config) do [timeout: config.timeout] end defp default_join_insert_all_opts(:upsert, join_keys, config) do [on_conflict: :nothing, conflict_target: join_keys, timeout: config.timeout] end # Association changes cannot pass through `insert_all/3`, so they are dropped from the row's # own upsert. (They are upserted separately, by recursing into each association's changesets.) defp drop_association_changes(%Ecto.Changeset{} = changeset, schema_module) do Map.update!(changeset, :changes, &Map.drop(&1, schema_module.__schema__(:associations))) end # Get all `has_many` and `has_one` associations for a given schema. defp get_schema_associations(schema_module, :has) do schema_module.__changeset__() |> Enum.filter(fn {_k, v} -> match?({:assoc, %Ecto.Association.Has{}}, v) end) |> Keyword.keys() end # Get all `many_to_many` associations for a given schema. defp get_schema_associations(schema_module, :many_to_many) do schema_module.__changeset__() |> Enum.filter(fn {_k, v} -> match?({:assoc, %Ecto.Association.ManyToMany{}}, v) end) |> Keyword.keys() end defp log_on_changeset_error(schema_module, changeset, verb) do item_id_or_ids = changeset_primary_key(schema_module, changeset) invalid_parent_attrs = changeset.errors |> Enum.reduce(%{}, fn {k, _v}, acc -> Map.put(acc, k, changeset.changes[k]) end) # If a parent has an error in an association, the error will appear as a changeset, which # clutters up the logs. So, remove association errors from the invalid attrs map. The error # message for the field will still appear in the logs, so the information about the error # will still get passed along |> Map.new(fn {k, v} -> if k in schema_module.__schema__(:associations), do: {k, :changesets_hidden_to_keep_logs_shorter}, else: {k, v} end) invalid_association_attrs = schema_module |> get_schema_associations(:has) # Only check associations that are present in the changeset's changes (i.e. they aren't nil) |> Enum.reject(&is_nil(changeset.changes[&1])) |> Enum.reduce(%{}, fn association, acc -> association_error_items = changeset.changes[association] # `has_one` changes are a single changeset; `has_many` changes are a list of changesets. |> List.wrap() |> Enum.reject(fn changeset -> Enum.empty?(changeset.errors) end) |> Enum.reduce([], fn changeset, acc -> changeset_error_items = changeset.errors |> Keyword.keys() |> Enum.reduce([], fn key, acc -> acc |> Keyword.put(key, changeset.changes[key]) end) changeset_error_items ++ acc end) if Enum.empty?(association_error_items), do: acc, else: acc |> Map.put(association, association_error_items) end) invalid_attrs = Map.merge(invalid_parent_attrs, invalid_association_attrs) Logger.debug( """ This changeset has one or more unrecoverable errors. The item associated with this \ changeset will not be #{verb_past_tense(verb)}.\ """, reason: changeset_error_reason(verb), schema_module: inspect(schema_module), item_id_or_ids: item_id_or_ids, # NOTE: If one item in an array contains an invalid value, the whole array will be logged fields_with_invalid_attrs: Map.keys(invalid_attrs), changeset_errors: changeset.errors ) end defp recover_changesets_with_recoverable_errors(changesets, recover_changeset_errors) when changesets == [] or recover_changeset_errors == %{} do changesets end defp recover_changesets_with_recoverable_errors(changesets, recover_changeset_errors) do Enum.map(changesets, &recover_changeset(&1, recover_changeset_errors)) end # Recover a single changeset, recursing into its nested association changesets first # (bottom-up). An association's error on the parent is cleared once all of that association's # child changesets are valid. The changeset itself is then recovered only if every remaining # error field has a fallback configured for the changeset's schema. defp recover_changeset(%Ecto.Changeset{valid?: true} = changeset, _recover_changeset_errors) do changeset end defp recover_changeset(changeset, recover_changeset_errors) do schema_module = changeset.data.__struct__ # Recover the nested association and embed changesets before the changeset's own errors, # since a nested error can only be cleared once all of its child changesets are valid changeset = (schema_module.__schema__(:associations) ++ schema_module.__schema__(:embeds)) |> Enum.reduce(changeset, fn association, acc_changeset -> recover_nested_changesets(acc_changeset, association, recover_changeset_errors) end) fallbacks = Map.get(recover_changeset_errors, schema_module, %{}) error_fields = changeset.errors |> Keyword.keys() |> Enum.uniq() # An error on an association or embed field itself (e.g. attrs that could not be cast) is # never recoverable: a fallback value would replace the field's changesets in the changes # with a bare value nested_fields = schema_module.__schema__(:associations) ++ schema_module.__schema__(:embeds) recoverable_field? = &(Map.has_key?(fallbacks, &1) and &1 not in nested_fields) cond do # An invalid child does not always leave an error entry on its parent (`cast_assoc/3` may # only set `valid?: false`), so the children's own validity is checked directly. A # changeset with an unrecovered child cannot be recovered not nested_changesets_valid?(changeset, schema_module) -> changeset error_fields == [] -> # Every error was an association error, and all child changesets have been recovered %{changeset | valid?: true} Enum.all?(error_fields, recoverable_field?) -> error_fields |> Enum.reduce(changeset, &recover_changeset_field(&2, &1, Map.fetch!(fallbacks, &1))) # Clear the changeset's errors and mark the changeset as valid |> Map.merge(%{errors: [], valid?: true}) true -> # The changeset has errors with no configured fallback. The changeset (or its parent, for # a nested changeset) will be removed later in the pipeline changeset end end # Check that every changeset in the association and embed changes is valid. An invalid child # does not always leave an error entry on its parent, so the parent's error list alone cannot # prove that all children have been recovered. defp nested_changesets_valid?(changeset, schema_module) do (schema_module.__schema__(:associations) ++ schema_module.__schema__(:embeds)) |> Enum.all?(fn association -> changeset.changes |> Map.get(association) |> List.wrap() |> Enum.all?(fn %Ecto.Changeset{} = child_changeset -> child_changeset.valid? _not_a_changeset -> true end) end) end # Recover the changesets in one association's (or embed's) changes, clearing the association's # error on the parent once every child changeset is valid. defp recover_nested_changesets(changeset, association, recover_changeset_errors) do case changeset.changes[association] do nil -> changeset children -> # `has_many`, `many_to_many`, and `embeds_many` changes are a list of changesets; # `has_one` and `embeds_one` changes are a single changeset. `List.wrap/1` normalizes # both into a list for recovery, and the original shape is restored when the changes are # updated recovered = children |> List.wrap() |> Enum.map(&recover_changeset(&1, recover_changeset_errors)) recovered_children = if is_list(children), do: recovered, else: hd(recovered) changeset = %{ changeset | changes: Map.put(changeset.changes, association, recovered_children) } if Enum.all?(recovered, & &1.valid?), do: Map.update!(changeset, :errors, &Keyword.delete(&1, association)), else: changeset end end defp recover_changeset_field(changeset, field, recover_to_value) do primary_key_info = changeset.data.__struct__.__schema__(:primary_key) |> Keyword.new(fn primary_key_field -> # The primary key may be absent from the changes (e.g. if the changeset function does # not require it), so avoid `Map.fetch!/2` {primary_key_field, Map.get(changeset.changes, primary_key_field)} end) Logger.debug(""" Recovered changeset error for struct #{Macro.to_string(changeset.data.__struct__)} with \ primary key(s) `#{inspect(primary_key_info)}` in the field `#{field}`.\ """) %{changeset | changes: Map.put(changeset.changes, field, recover_to_value)} end # One `:warning` per call summarizes every skipped item, accumulated across all chunks. The # per-item details are logged at the `:debug` level, so a large batch of invalid rows cannot # flood the log. defp log_skipped_changesets_summary(schema_module, totals, verb) do truncation_note = if totals.skipped > @skipped_item_ids_log_limit, do: " The first #{@skipped_item_ids_log_limit} skipped item IDs are listed.", else: "" Logger.warning( """ Skipped #{totals.skipped} of #{totals.written + totals.skipped} items because their \ changesets had unrecoverable errors. The skipped items were not \ #{verb_past_tense(verb)}. Details for each skipped item are logged at the `:debug` \ level.#{truncation_note}\ """, reason: items_skipped_reason(verb), schema_module: inspect(schema_module), skipped_count: totals.skipped, item_ids: totals.skipped_item_ids ) end # Log metadata `:reason` atoms, per verb defp changeset_error_reason(:insert), do: :insert_changeset_error defp changeset_error_reason(:upsert), do: :upsert_changeset_error defp items_skipped_reason(:insert), do: :insert_items_skipped defp items_skipped_reason(:upsert), do: :upsert_items_skipped end