Bulk insert for Ecto via unnest(...).

A plain Ecto.Repo.insert_all/3 builds VALUES ($1,$2),($3,$4),..., so the SQL text grows with the number of rows — every batch size is a different prepared statement. PgBouncer (transaction mode) dislikes that, and Postgres caps parameters at ~65535.

This library generates constant SQL text, independent of the row count:

INSERT INTO "events" ("type","user_id")
(SELECT f0."type", f0."user_id"
 FROM (SELECT * FROM unnest($1::text[], $2::bigint[]) AS u("type","user_id")) AS f0)

Whether you insert 1 or 10,000 rows, the statement is identical. The query is assembled from Ecto building blocks (fragment/dynamic) and handed to Ecto.Repo.insert_all/3, which renders ON CONFLICT/RETURNING/prefix and loads structs natively.

API

Two disjoint maps:

• a columns map %{col => list} — each column goes into unnest as an array,
• :placeholders %{col => value} — constants broadcast onto every row.

EctoUnnest.insert_all(Repo, Event,
  %{user_id: [1, 2, 3], type: ["click", "view", "click"]},
  placeholders: %{inserted_at: ~U[2026-06-17 10:00:00Z]},
  returning: true
)

Column names, table names, schema prefixes, conflict targets, and type overrides are SQL structure. They should come from trusted application code or configuration, not user input. Put user-controlled data in the column lists or :placeholders so Ecto/Postgrex can send it as parameters.

Options (same as Ecto.Repo.insert_all/3)

• :placeholders — %{col => value} of constant columns (default %{})

• :returning — true | false | [field] (default false)

• :prefix — schema prefix (overrides @schema_prefix)

• :on_conflict — :raise | :nothing | :replace_all | {:replace, fields} | {:replace_all_except, fields} | [set: kw, inc: kw] | Ecto.Query.t(). A full query enables a conditional DO UPDATE ... WHERE ... (see "Conflicts" below)

• :conflict_target — [col] | {:unsafe_fragment, binary}

• :types — %{col => pg_type} override for type inference, where pg_type is an atom (recommended — assumed app-controlled, so rendered straight into the SQL cast) or a string. For a placeholder it becomes a raw ::pg_type cast, so it can name a custom PG type (e.g. a domain :kafka_topic_name). The value is rendered into SQL verbatim, so it is fail-closed: Ecto's default PG types (bigint, text, jsonb, timestamptz, …) are always accepted, but any other spelling (a domain, an alias like int4, a modifier like numeric(10,2)) must be vouched for in config, or it raises:

  config :ecto_unnest, :allowed_types, [:int4, "kafka_topic_name"]

• :json — [col] columns to force into JSON mode (see "JSON columns" below)

• :cache_statement — prepared-statement cache name. Defaults to "ecto_unnest_all_#{table}_#{arity}", where arity is the number of unnest (array) columns, so distinct column shapes for one table do not share a cache slot (see "Prepared-statement cache" below). Pass a binary to override, or nil to fall back to Ecto's "ecto_insert_all_#{table}"

• :require_all_fields — true to assert every schema field is accounted for (see "Completeness check" below; schema sources only). Defaults to the config :ecto_unnest, :require_all_fields value, else false

Completeness check

With require_all_fields: true every field of the schema must appear either in the columns map or in :placeholders, otherwise it raises listing what is missing. This catches a column you forgot — without it a missing field silently falls back to its DB default (or NULL). The autogenerated primary key and timestamps() fields are exempt (Ecto/the DB fills them). A field you mean to leave at its default goes in :placeholders as nil:

EctoUnnest.insert_all(Repo, Event,
  %{user_id: [1, 2], type: ["a", "b"], score: [1.0, 2.0], tags: [nil, nil]},
  placeholders: %{inserted_at: ts, payload: nil},
  require_all_fields: true)

Only valid for a schema source (a binary source has no field list and raises).

To make it a test-only safety net (off in production), skip the per-call option and set it in config for that environment:

# config/test.exs
config :ecto_unnest, :require_all_fields, true

An explicit :require_all_fields in opts always overrides the config value.

JSON columns

A per-row value destined for a jsonb column (Ecto type :map, {:map, _} or {:array, :map}) cannot ride in unnest as a ::jsonb[] array — unnest flattens multi-dimensional arrays and Postgrex would double-encode. Such columns are shipped instead as a 1-D text[] of pre-encoded JSON with a ::jsonb cast in the projection. Schema sources detect this automatically from the Ecto type; on binary sources use types: %{col: :jsonb} or json: [:col]. Raw terms are JSON encoded with the configured :postgrex :json_library; already-encoded strings pass through untouched.

EctoUnnest.insert_all(Repo, Doc,
  %{id: [id1, id2], summary: [[%{"a" => 1}], [%{"b" => 2}]]},
  placeholders: %{created_at: ts})

Prepared-statement cache

Postgrex caches prepared statements under a name. Ecto's insert_all derives it from the table alone ("ecto_insert_all_#{table}"), which is fine when the SQL for a table never changes. But here the SQL depends on the column shape: each distinct set of unnest (array) columns renders a different statement, so sharing one cache name per table would make Postgrex re-prepare every time the shape changes. The default name therefore appends the unnest arity:

ecto_unnest_all_events_2   # 2 unnest columns
ecto_unnest_all_events_3   # 3 unnest columns

The row count never enters the name (the SQL is constant across it — that is the whole point). Arity does not distinguish two same-arity shapes with different column names; if you insert such shapes into one table, give each an explicit :cache_statement. Pass cache_statement: nil to restore Ecto's default.

Conflicts

Besides the keyword form, :on_conflict accepts a full Ecto.Query (as Ecto.Repo.insert_all/3 does) for a conditional update:

from(s in Setting, update: [set: [deleted_at: nil]], where: not is_nil(s.deleted_at))

Reading: virtual table

table/3 exposes the same unnest(...) source as a composable %Ecto.Query{}, so you can SELECT from it or join it into an UPDATE. See table/3.

Syncing a table from a file

sync_all/4 makes a table match a source — upsert every row, then delete the rows the source no longer has. That second half is what insert_all/4 cannot express: ON CONFLICT only ever sees the rows you sent, so a row you deleted from a YAML or CSV file in git is a row Postgres never hears about.

EctoUnnest.sync_all(Repo, Setting, %{key: ["a", "b"], value: ["1", "2"]})
#=> %{upserted: 2, deleted: 1, ...}

Both statements keep text independent of the row count, and deleting is treated as the dangerous half it is (:on_delete guards against a silent CASCADE). See sync_all/4.

Limitations

• non-JSON array-typed columns ({:array, _} other than {:array, :map}) that vary per row are unsupported (unnest flattens multi-dimensional arrays) — we raise a clear error. JSON arrays ({:array, :map}) are supported via JSON mode (see "JSON columns"),
• binary sources ("table") require :types, and :returning as a field list (no __schema__).