Keyset pagination for Ecto queries.

A cursor holds a sort order and the position of the last row read. Each page is fetched with a WHERE clause that continues right after that position, so pages stay consistent when rows are inserted or deleted between calls. With an index matching the sort order, the database seeks to the position in the index, so deep pages stay fast.

Create a cursor with a sort order and a page size, then fetch pages with a base query and a repo. Each page carries the cursor for the next one.

cursor = Ark.EctoCursor.new(order_by: [desc: :inserted_at, asc: :id], page_size: 20)

%Ark.EctoCursor.Page{entries: posts, next_cursor: next} =
  Ark.EctoCursor.fetch_page(cursor, MyApp.Post, MyApp.Repo)

%Ark.EctoCursor.Page{entries: more_posts} =
  Ark.EctoCursor.fetch_page(next, MyApp.Post, MyApp.Repo)

The :has_next_page field of Ark.EctoCursor.Page tells whether more rows follow the page.

To go through every row, stream/4 fetches the pages lazily as the stream is consumed.

Ark.EctoCursor.new(order_by: :id, page_size: 500)
|> Ark.EctoCursor.stream(MyApp.Post, MyApp.Repo)
|> Enum.each(&MyApp.Search.index_post/1)

Sort order

The :order_by option lists the fields that define the order of the rows. A plain field name sorts in ascending order, and {:asc, field} or {:desc, field} give an explicit direction, as in Ecto.Query.order_by/3.

Choose the sort fields with these rules in mind:

• Together, the fields must identify a single row. End the list with a unique field, usually the primary key, to break ties between rows that share the other values. For instance [desc: :inserted_at, asc: :id].
• The fields must be non-nullable columns. Rows with a NULL sort value can be skipped from the pages, and a row with a nil sort value raises an ArgumentError when it becomes the position of the next cursor.
• An index on the same fields, in the same order and directions (or with all directions reversed), lets the database read each page directly from the index. For the example above, that is create index(:posts, ["inserted_at DESC", :id]) in a migration.

With several sort fields, the query also bounds the first field with a redundant >= (or <=) condition, which lets query planners seek the index from the position. The :use_index option of fetch_page/4, stream/4 and query/3 controls this behaviour.

Base queries

The base query can be any Ecto.Queryable: a schema, a {source, schema} tuple, or a query with where, join, select and distinct: true clauses. The cursor adds the order_by, the limit, and the conditions on the position, which shapes the base query:

• The base query leaves out limit, offset and order_by. The limit comes from the :page_size option, and the order from the cursor.
• Combinations such as Ecto.Query.union/2 go inside a subquery: from(s in subquery(union_all(q1, ^q2))).
• distinct uses true rather than a list of expressions.

Each row returned by the query must be a map or a struct that holds every sort field, under an atom or string key. Rows selected as schema structs or with Ecto.Query.API.map/2 meet this naturally. With a custom select, include the sort fields in the selected map.

By default, sort fields are read from the root source of the query. To sort on a joined source, give the cursor a :binding naming a binding of the query, defined with the :as option. A joined binding needs an explicit select that returns the sort fields of that binding, under their field names. This example lists the members of a group, sorted by user name.

query =
  from m in MyApp.Membership,
    where: m.group_id == ^group_id,
    join: u in assoc(m, :user),
    as: :user,
    select: %{id: u.id, name: u.name, role: m.role}

cursor = Ark.EctoCursor.new(order_by: [:name, :id], page_size: 20, binding: :user)

Persisting cursors

A cursor is an embedded schema, so it can be stored in a column of a parent schema with Ecto.Schema.embeds_one/3. This is useful to resume a long running job, or to keep polling a table for new rows. The cursor of the last page still points after the last row, so with a sort order that follows insertion, such as :id or :inserted_at, later calls return the rows inserted since then. Concurrent transactions can commit rows out of that order though, and a row committed after a later row was read is skipped.

defmodule MyApp.Export do
  use Ecto.Schema

  schema "exports" do
    embeds_one :cursor, Ark.EctoCursor, on_replace: :delete
  end
end

The database column holds a map, declared with add :cursor, :map in the migration that creates the parent table.

Only the sort order and the position are stored, so resume!/2 turns a stored cursor back into a working one, with the page size and binding given again. It also accepts nil, which starts a new cursor for a parent that has none yet.

@cursor_opts [order_by: [asc: :inserted_at, asc: :id], page_size: 500]

def export_next_batch(%MyApp.Export{cursor: stored} = export) do
  cursor = Ark.EctoCursor.resume!(stored, @cursor_opts)

  %Ark.EctoCursor.Page{entries: posts, next_cursor: next} =
    Ark.EctoCursor.fetch_page(cursor, MyApp.Post, MyApp.Repo)

  :ok = MyApp.Storage.write_posts(posts)

  export
  |> Ecto.Changeset.change()
  |> Ecto.Changeset.put_embed(:cursor, next)
  |> MyApp.Repo.update!()
end

Stored as JSON, position values take their JSON form, a datetime becomes a string for instance. Ecto casts them back to the type of the sort fields when building the query, using the schema of the queried source. Use a base query with a schema, such as MyApp.Post or {"posts", MyApp.Post}, with cursors loaded from storage.

Ecto.embedded_dump/2 and Ecto.embedded_load/3 convert a cursor to and from JSON data for storage outside of a schema.