EctoUnnest (ecto_unnest v0.3.2)

Copy Markdown View Source

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__).

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

columns()

@type columns() :: %{required(atom()) => list()}

source()

@type source() :: module() | binary()

sync_result()

@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

insert_all(repo, schema, columns, opts \\ [])

@spec insert_all(Ecto.Repo.t(), source(), columns(), keyword()) ::
  {non_neg_integer(), [struct()] | nil}

sync_all(repo, schema, columns, opts \\ [])

@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 — true to 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 — true to 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 is ON DELETE CASCADE/SET NULL/SET DEFAULT, since those change child rows with no error. NO ACTION/RESTRICT children need no check: Postgres raises foreign_key_violation and 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} — set field to 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).

table(schema, columns, opts \\ [])

@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([])

to_sql(schema, columns, opts \\ [])

@spec to_sql(source(), columns(), keyword()) :: {String.t(), [term()]}

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).