Multiple Ecto transactions give timeout

hudsonbay

I’m in a situation where I have to insert 9000 registries in bulk into the database with Ecto in PostgreSQL.

I already have a dedicated function for inserting an array of elements BulkOperations.bulk_create(MyStruct, list). It’s basically an Ecto.Multi, so everything in the list must be inserted because it’s a transaction.

No problem with that. If I insert a list of 10 elements everything’s OK. But if the list has 9000 registries we have a problem, because I receive a timeout when the connection stays open for more than 15000 ms doing the operations. I mean, 9000 registries , it takes a lot.

Now, I decided to separate everything by chunks. Like this:

        total_rows_affected =
          list
          |> Enum.chunk_every(20)
          |> Enum.map(&BulkOperations.bulk_create(MyStruct, &1))
          |> Enum.reduce(0, fn %{rows_affected: rows_affected}, acc ->
            rows_affected + acc
          end)

        %{errors: [], rows_affected: total_rows_affected}

It works for for 9000 registries.

But I feel it’s not secure because if I insert a chunk of elements with &BulkOperations.bulk_create(MyStruct, &1) and that operation is not succesful (because of a database key conflict maybe) then I have a problem because some operations will fail and some others don’t.

My solution was to create a transaction, so everything has to be successful. Like this:

    {:ok, result} =
      Repo.transaction(fn ->
        total_rows_affected =
          list
          |> Enum.chunk_every(10)
          |> Enum.map(&BulkOperations.bulk_create(MyStruct, &1))
          |> Enum.reduce(0, fn %{rows_affected: rows_affected}, acc ->
            rows_affected + acc
          end)

        %{errors: [], rows_affected: total_rows_affected}
      end)

    result

And it works but I receive a timeout when I try to insert 9000 registries because I believe everything is using the same connection opened by Repo.transaction.

So, my question is, how can I insert all of this 9000 registries but making sure that everything is inserted?

11 comments

#ecto #ecto_multi #transactions #timeout

0 3874 11

2021-12-14 23:46:13 UTC

Showing Posts 1 to 10

Show Best Posts
Show All (oldest first)
Show All (newest first)

RudManusachi

Have you tried to pass timeout: :infinity option to Repo.insert_all/3 or Repo.transaction/2?

Post #1

hudsonbay

No, I haven’t. Because I’m worried that if I do that I could overload the server’s memory were the database is hosted. Because it’s too much to insert. Too much work

Post #2

axelson

Scenic Core Team

Are you using insert_all? If you’re not then you could get a speed up that way.

Post #3

dimitarvp

Why don’t you try before it so you know for sure?

Post #4

odix67

I think Jason’s proposal is the best option, btw. honestly, I don’t know of any database that can span a transaction across multiple connections, but I can be wrong.

Post #5

hudsonbay

Actually, no. I’m not using an insert_all function. It’s a Ecto.Multi.insert one when calling BulkOperations.bulk_create/2

This is what I did following your advices

{:ok, result} =
    Repo.transaction(fn ->
      total_rows_affected =
        list
        |> Enum.chunk_every(10)
        |> Enum.map(&BulkOperations.bulk_create(MyStruct, &1))
        |> Enum.reduce(0, fn %{rows_affected: rows_affected}, acc ->
          rows_affected + acc
        end)

      %{errors: [], rows_affected: total_rows_affected}
    end, [{:timeout, :infinity}])   # <============= this is what I added

  result

But I received :

"2021-12-13 19:29:43.254 UTC [1299724]: [1-1] db=my_db,user=postgres LOG:  unexpected EOF on client connection with an open transaction"

I don’t know why the client(Phoenix) is closing the connection because I set this to :infinity

I also did it this way: timeout: :infinity and Phoenix is closing the connection after some time

Post #6

dimitarvp

What does bulk_create do exactly?

Post #7

hudsonbay

@dimitarvp it was inserting one by one. I changed it with the help of a friend and it worked. insert_all does the job as @axelson was saying. Now it looks like this:

def bulk_create(entity, items) do
    Ecto.Multi.new()
    |> Ecto.Multi.insert_all(:insert_all, entity, items, on_conflict: :nothing)
    |> Repo.transaction()
    |> case do
      {:ok, _} ->
        %{rows_affected: length(items), errors: []}

      {
        :error,
        _,
        %Ecto.Changeset{
          changes: changes,
          errors: errors
        },
        _
      } ->
        %{
          rows_affected: 0,
          errors: "Changes #{changes} couldn't be created because #{inspect(errors)}"
        }
    end
  end

But the problem is that I had to take care of timestamps because insert_all doesn’t insert them. So a migration to set a default value of now() solves the issue.

So, issue solved but I still won’t know why it was closing the connection even when I set the timeout to :infinity with the previous function. BTW, now with the insert_all solution I can stick to the default timeout of 15_000ms

Post #8

dimitarvp

If memory serves, the second element in the :ok tuple should already contain the number of items affected.

Post #9

hudsonbay

Yes, but as it’s a transaction, all of the initial items in the list should be affected. That’s why I used items param with length(items). Makes sense to you now? Thanks for noticing that, BTW

Post #10

Load more posts (1 remaining)