Ecto.Repo.insert_all/3 and Ecto.Multi.insert_all/5 support? about oban HOT 5 CLOSED

Something similar to this came up in the recipe post that I'm working on: how do you insert a lot of jobs (where I'd define a lot as "too many to simply loop over") at once and retain features like uniqueness and prefix support.

This seems like a great solution to the problem, though I do have an open questions about your proposal:

Is the idea that new_for_insert_all/1 would return a map rather than a changeset, the way insert_all/3 functions expect data? That poses an issue because the unique field on jobs is virtual, so it is dropped when we convert the changeset to a map for insert. I think Oban.insert_all/2 should receive a list of changesets instead. It isn't as consistent with Repo.insert_all/3, but it does match Oban.insert/2.

from oban.

Oban knows its Repo, so I think that a list of changesets or Job structs is the best choice. I am less concerned about uniqueness support for the case where I’d be using an insert_all, although other users may be more concerned about that. My use case is something like:

Multi.new()
|> Multi.insert_all("deactivate_accounts", expired_customers, on_conflict: update_clause)
|> Multi.insert_all("deactivate_memberships", expired_customers, on_conflict: update_clause)
|> Oban.insert_all("notify_customers", Enum.map(expired_customers, &Worker.new/1))
|> Repo.transaction()

In some ways this is an abuse of insert_all (there are no new records that will be created, guaranteed), but update_all is too limited in ability. We’re guaranteed, for our particular use case, that every expired customer will be notified because we’re guaranteed that they will all be deactivated.

from oban.

@halostatue Support for insert_all has been added. You may be interested in the new Oban.Job.to_map/1 function as well, if using Oban.insert_all/2,4 isn't sufficient.

This will be released in 0.9.0 today or tomorrow 👍

from oban.

Thank you!

from oban.

Fun little benchmark demonstrating the relative improvement from batch inserting jobs.

Name                    ips        average  deviation         median         99th %
insert_all            22.12       45.22 ms    ±18.31%       43.96 ms       75.43 ms
insert                 1.75      570.56 ms     ±3.08%      570.52 ms      640.43 ms
insert_unique          0.99     1014.89 ms     ±0.73%     1015.18 ms     1038.70 ms

Comparison:
insert_all            22.12
insert                 1.75 - 12.62x slower +525.34 ms
insert_unique          0.99 - 22.44x slower +969.67 ms

Here is the benchee script itself:

defmodule WorkerA do
  use Oban.Worker, queue: :bench

  def perform(_, _), do: :ok
end

defmodule WorkerB do
  use Oban.Worker, queue: :unique, unique: [period: 60]

  def perform(_, _), do: :ok
end

Logger.configure(level: :warn)
Oban.Test.Repo.start_link()
Oban.start_link(repo: Oban.Test.Repo, queues: false)

insert = fn ->
  Enum.each(1..1_000, fn index ->
    %{id: index}
    |> WorkerA.new()
    |> Oban.insert()
  end)
end

insert_all = fn ->
  1..1_000
  |> Enum.map(&WorkerA.new(%{id: &1}))
  |> Oban.insert_all()
end

insert_unique = fn ->
  Enum.each(1..1000, fn index ->
    %{id: index}
    |> WorkerB.new()
    |> Oban.insert()
  end)
end

Benchee.run(
  %{"insert" => insert,
    "insert_all" => insert_all,
    "insert_unique" => insert_unique
  },
  time: 30,
  warmup: 1
)

from oban.