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 intounnestas 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](defaultfalse):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 conditionalDO UPDATE ... WHERE ...(see "Conflicts" below):conflict_target—[col] | {:unsafe_fragment, binary}:types—%{col => pg_type}override for type inference, wherepg_typeis an atom (recommended — assumed app-controlled, so rendered straight into the SQL cast) or a string. For a placeholder it becomes a raw::pg_typecast, 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 likeint4, a modifier likenumeric(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}", wherearityis the number ofunnest(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, ornilto fall back to Ecto's"ecto_insert_all_#{table}":require_all_fields—trueto assert every schema field is accounted for (see "Completeness check" below; schema sources only). Defaults to theconfig :ecto_unnest, :require_all_fieldsvalue, elsefalse
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, trueAn 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 columnsThe 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 (unnestflattens 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:returningas a field list (no__schema__).
Summary
Functions
Makes the table match the source: upsert every row, then remove the rows the source no longer has.
A virtual table built from unnest(...) as a composable %Ecto.Query{}.
Returns {sql, params} without executing the query.
Types
@type sync_result() :: %{ upserted: non_neg_integer(), deleted: non_neg_integer(), soft_deleted: non_neg_integer(), deleted_keys: [map()], skipped_keys: [map()], on_delete: term() }
What sync_all/4 did.
:upserted— rows inserted or updated (Postgres does not distinguish them),:deleted— rows hard-deleted because the source no longer has them,:soft_deleted— rows tombstoned instead of deleted ({:soft, field}),:deleted_keys— the key of every row removed (hard or soft), as a map,:skipped_keys— absent rows deliberately left in place (:skip_referenced),:on_delete— the normalized mode that ran.
Functions
@spec insert_all(Ecto.Repo.t(), source(), columns(), keyword()) :: {non_neg_integer(), [struct()] | nil}
@spec sync_all(Ecto.Repo.t(), source(), columns(), keyword()) :: sync_result()
Makes the table match the source: upsert every row, then remove the rows the source no longer has.
This is the operation you want when a file in git — YAML, CSV, JSON — is the
source of truth for a table. insert_all/4 alone cannot express it: ON CONFLICT
only ever sees the rows you sent, so a row you deleted from the file is a row
Postgres never hears about.
EctoUnnest.sync_all(Repo, Setting,
%{key: ["a", "b"], value: ["1", "2"]},
key: [:key])
#=> %{upserted: 2, deleted: 1, soft_deleted: 0,
#=> deleted_keys: [%{key: "c"}], skipped_keys: [], on_delete: :restrict}Both statements run in one transaction, and both have SQL text independent of the
row count (the delete's NOT EXISTS reads from the same unnest(...) source), so
the prepared-statement story of insert_all/4 carries over.
Options
Everything insert_all/4 takes, plus:
:key— the column(s) identifying a row, as a list. Defaults to the schema's primary key. Also the default:conflict_target. Must be present in the columns map (the file has to say which row it is describing),:on_delete— what to do with a row the source no longer has (below), default:restrict,:allow_empty—trueto permit a source with no rows. An empty source means "delete everything", which is almost always a parse failure rather than an intention, so it raises by default,:lock—trueto serialize concurrent syncs of this table (below).
Concurrent syncs
A sync is two statements, so two of them running at once can interleave: both
upsert, then both delete, and the loser's rows are removed by the winner's delete
before its own upsert is visible. lock: true takes a transaction-scoped advisory
lock keyed on the table's oid, so the second sync waits for the first:
EctoUnnest.sync_all(Repo, Setting, cols, lock: true)It is off by default because it blocks — one extra statement
(pg_advisory_xact_lock) and a wait on whoever holds it. Turn it on when a sync
can be triggered more than once at a time (a deploy hook that may overlap, a
scheduled job, a web endpoint). The lock is released when the transaction ends,
which for a nested call means the outermost transaction.
Pass an integer instead of true to choose the key yourself, so other code can
take the same lock:
EctoUnnest.sync_all(Repo, Setting, cols, lock: 8_675_309)Advisory locks share one key space per database: a bare true derives the key
from the table oid, which is unique per table but could in principle collide with
another library's key. Pass your own integer if that matters.
:conflict_target defaults to :key, and :on_conflict to replacing exactly the
columns the source carries — not Ecto's :replace_all. Pass either to override;
a full Ecto.Query :on_conflict gives you a conditional DO UPDATE ... WHERE,
so untouched rows do not churn.
Row identity
A row keeps its surrogate primary key across syncs. Syncing on a natural key is the normal case:
EctoUnnest.sync_all(Repo, Event, %{user_id: [1, 2, 3], type: ["a", "b", "c"]},
key: [:user_id]):id is not in the source, so it is not in the INSERT's column list and not in
the DO UPDATE SET — existing rows keep the id they had, and deleting a row does
not renumber the others.
This is also why the default :on_conflict is not Ecto's :replace_all, which
expands to __schema__(:updatable_fields) — primary key included. That renders
SET "id" = EXCLUDED."id", and for a column the INSERT does not list EXCLUDED
holds the column's default: NULL, or the next sequence value for a serial. Every
sync would renumber every row it touched and repoint anything referencing them.
So a caller-supplied :on_conflict is checked: if its replace list names a column
the source does not provide, it raises rather than resetting it. Add the column to
the source, exclude it ({:replace_all_except, [:id]}), or leave :on_conflict
alone.
:on_delete
Deletion is the only part of a sync that can destroy data the source never
described, because a CASCADE reaches rows in other tables.
:restrict(default) — refuse to run if any foreign key pointing at this table isON DELETE CASCADE/SET NULL/SET DEFAULT, since those change child rows with no error.NO ACTION/RESTRICTchildren need no check: Postgres raisesforeign_key_violationand the transaction rolls the whole sync back,:skip_referenced— delete only the absent rows nothing references. The referencing tables and columns come from the catalog, so composite keys work; the rows left behind are reported in:skipped_keys,{:soft, field}— setfieldto the current time instead of deleting, which sidesteps foreign keys entirely and is usually the right answer for git-tracked reference data. The value is derived from the field's Ecto type; pass{:soft, field, value}to choose it yourself. Rows already tombstoned are left alone,:cascade— no catalog check, plain delete. Destructive foreign keys fire. Explicit opt-in,:nothing— upsert only, delete nothing.
Only :restrict and :skip_referenced read the catalog (one small query each).
@spec table(source(), columns(), keyword()) :: Ecto.Query.t()
A virtual table built from unnest(...) as a composable %Ecto.Query{}.
Each column %{col => list} becomes an unnest column. The result can be used
like any Ecto source — where, order_by, select, Repo.all/2, and through
subquery/1 also in a join for update_all/delete_all.
Types come from the schema (like insert_all/4) or from :types for sources
without a schema. The binding is named :s by default (:as option).
q = EctoUnnest.table(Event, %{user_id: [1, 2, 3], type: ["a", "b", "c"]})
from([s: s] in q, where: s.user_id > 1, select: {s.user_id, s.type})
|> Repo.all()For a join, wrap it in subquery/1 (which carries the parameters) and give it a
select:
src = from([s: s] in q, select: %{user_id: s.user_id, type: s.type})
from(e in Event, join: s in subquery(src), on: e.user_id == s.user_id,
update: [set: [type: s.type]])
|> Repo.update_all([])
Returns {sql, params} without executing the query.
Useful for debugging and for tests with no database connection — the whole plan
build (type inference) is pure, and the text itself is rendered by the Postgres
adapter's Connection module (the same building blocks Repo.insert_all/3 uses).