Ecto multiple streams in 1 transaction

I know it might not be the solution you’re looking for, but instead of two separate queries you could use subqueries to make them a single, but nested query. No need to send the results of the first query back to elixir in the first place.

Given the the final objective is to perform a deletion, and that POSTGRES SQL has a limitation for deletes that basically forces me to perform 2 queries (the first where I get the ids, and the second where I delete the users) I don’t see how I could implement your solution.

Could you elaborate more? I am genuinely curious.

The SQL for this would be:

DELETE FROM users u
USING purchases p
WHERE u.purchase_id = p.id and p.item_id = $1

In Ecto you can do this as simply as:

query = from u in Users,
  join: p in assoc(u, :purchase),
  where: p.item_id == ^item_id

Repo.delete_all(query)

What @benwilson512 suggested is even better than a subquery, but for more information on subqueries see:

But wont this still blow up if the results dont fit into memory ?
Or is Repo.delete_all free from memory?

What results are you expecting from delete_all?

The results do not go to ecto at all, it runs a DELETE on the postgres side. Postgres is tuned to deal with essentially arbitrary amounts of data through a mix of temporary working spaces in both systems ram and disk. A properly tuned postgres will not run out of memory no matter how many records you are deleting.

It is my understanding that if possible delete_all will return {non_neg_integer(), nil | [term()]}, where term is the deleted item, if the DB beneath supports it, which in my case does.

Am I missing something there?

@benwilson512 Comparing this solution to another solution using streams, which one would you pick?
For the sake of this post, I will copy a solution from SO:

Repo.transacion(fn ->
  max_rows = 500

  purchase_ids
  |> Repo.stream(max_rows: max_rows)
  |> Stream.chunk_every(max_rows)
  |> Stream.each(fn ids ->
     Users
     |> where([u], u.purchase_id in ^ids)
     |> Repo.delete_all()
  end)
  |> Stream.run()
end, timeout: :infinity)

The original post can be found:

The streaming solution sounds pretty awesome to me, and given both do the same, which one would have more resilience to failure?
(I know that transactions hold a connection, but I am not near the connection limit)

What you’re missing is that it is only returned if you ask it to be returned. By default, it returns nothing at all except a count of the entities deleted.

In regards to the query:

query = from u in Users,
  join: p in assoc(u, :purchase),
  where: p.item_id == ^item_id

Repo.delete_all(query)

I get the error:

** (Ecto.QueryError) iex:41: could not find association `purchase` on schema Users in query:
...

Is there some alteration I need to do to the schema of Users or Purchase?
The SQL code did work though, so I am having a hard trouble understanding why the Ecto one doesn’t. I am assuming this is because I have an issue in some schema?

Also, I cannot use:

query = from u in Users,
  join: p in assoc(u, Purchase),
  where: p.item_id == ^item_id

even though Purchase is aliased, as this will give me a different error (basically I understand this only works with atoms).