EctoUnnest (ecto_unnest v0.2.0)

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
)

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)

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

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.

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

Types

columns()
source()

Functions

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

A virtual table built from unnest(...) as a composable %Ecto.Query{}.

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

Returns {sql, params} without executing the query.

Types

columns()

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

source()

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

Functions

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

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

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