mirror of
https://github.com/philomena-dev/philomena.git
synced 2024-11-23 20:18:00 +01:00
129 lines
3.9 KiB
Elixir
129 lines
3.9 KiB
Elixir
defmodule PhilomenaQuery.Batch do
|
|
@moduledoc """
|
|
Locking-reduced database batch operations.
|
|
|
|
These operations are non-transactional by their very nature. This prevents inadvertent
|
|
downtimes due to blocking, but can result in consistency errors in the database,
|
|
and so should be used sparingly.
|
|
|
|
They are best suited for when large numbers of rows can be expected to be processed,
|
|
as doing so may otherwise result in Ecto timing out the query.
|
|
"""
|
|
|
|
alias Philomena.Repo
|
|
import Ecto.Query
|
|
|
|
@typedoc """
|
|
Represents an object which may be operated on via `m:Ecto.Query`.
|
|
|
|
This could be a schema object (e.g. `m:Philomena.Images.Image`) or a fully formed query
|
|
`from i in Image, where: i.hidden_from_users == false`.
|
|
"""
|
|
@type queryable :: any()
|
|
|
|
@type batch_size :: {:batch_size, integer()}
|
|
@type id_field :: {:id_field, atom()}
|
|
@type batch_options :: [batch_size() | id_field()]
|
|
|
|
@typedoc """
|
|
The callback for `record_batches/3`.
|
|
|
|
Takes a list of schema structs which were returned in the batch. Return value is ignored.
|
|
"""
|
|
@type record_batch_callback :: ([struct()] -> any())
|
|
|
|
@typedoc """
|
|
The callback for `query_batches/3`.
|
|
|
|
Takes an `m:Ecto.Query` that can be processed with `m:Philomena.Repo` query commands, such
|
|
as `Philomena.Repo.update_all/3` or `Philomena.Repo.delete_all/2`. Return value is ignored.
|
|
"""
|
|
@type query_batch_callback :: ([Ecto.Query.t()] -> any())
|
|
|
|
@doc """
|
|
Execute a callback with lists of schema structures on a queryable,
|
|
using batches to avoid locking.
|
|
|
|
Valid options:
|
|
* `batch_size` (integer) - the number of records to load per batch
|
|
* `id_field` (atom) - the name of the field containing the ID
|
|
|
|
## Example
|
|
|
|
queryable = from i in Image, where: i.image_width >= 1920
|
|
|
|
cb = fn images ->
|
|
Enum.each(images, &IO.inspect(&1.id))
|
|
end
|
|
|
|
PhilomenaQuery.Batch.record_batches(queryable, cb)
|
|
|
|
"""
|
|
@spec record_batches(queryable(), batch_options(), record_batch_callback()) :: []
|
|
def record_batches(queryable, opts \\ [], callback) do
|
|
query_batches(queryable, opts, &callback.(Repo.all(&1)))
|
|
end
|
|
|
|
@doc """
|
|
Execute a callback with bulk queries on a queryable, using batches to avoid locking.
|
|
|
|
Valid options:
|
|
* `batch_size` (integer) - the number of records to load per batch
|
|
* `id_field` (atom) - the name of the field containing the ID
|
|
|
|
> #### Info {: .info}
|
|
>
|
|
> If you are looking to receive schema structures (e.g., you are querying for `Image`s,
|
|
> and you want to receive `Image` objects, then use `record_batches/3` instead.
|
|
|
|
An `m:Ecto.Query` which selects all IDs in the current batch is passed into the callback
|
|
during each invocation.
|
|
|
|
## Example
|
|
|
|
queryable = from ui in ImageVote, where: ui.user_id == 1234
|
|
|
|
opts = [id_field: :image_id]
|
|
|
|
cb = fn bulk_query ->
|
|
Repo.delete_all(bulk_query)
|
|
end
|
|
|
|
PhilomenaQuery.Batch.query_batches(queryable, opts, cb)
|
|
|
|
"""
|
|
@spec query_batches(queryable(), batch_options(), query_batch_callback()) :: []
|
|
def query_batches(queryable, opts \\ [], callback) do
|
|
ids = load_ids(queryable, -1, opts)
|
|
|
|
query_batches(queryable, opts, callback, ids)
|
|
end
|
|
|
|
defp query_batches(_queryable, _opts, _callback, []), do: []
|
|
|
|
defp query_batches(queryable, opts, callback, ids) do
|
|
id_field = Keyword.get(opts, :id_field, :id)
|
|
|
|
queryable
|
|
|> where([m], field(m, ^id_field) in ^ids)
|
|
|> callback.()
|
|
|
|
ids = load_ids(queryable, Enum.max(ids), opts)
|
|
|
|
query_batches(queryable, opts, callback, ids)
|
|
end
|
|
|
|
defp load_ids(queryable, max_id, opts) do
|
|
id_field = Keyword.get(opts, :id_field, :id)
|
|
batch_size = Keyword.get(opts, :batch_size, 1000)
|
|
|
|
queryable
|
|
|> exclude(:preload)
|
|
|> exclude(:order_by)
|
|
|> order_by(asc: ^id_field)
|
|
|> where([m], field(m, ^id_field) > ^max_id)
|
|
|> select([m], field(m, ^id_field))
|
|
|> limit(^batch_size)
|
|
|> Repo.all()
|
|
end
|
|
end
|