You can increase the timeout for the transaction where streaming happens:
Repo.transaction(fn -> ... end, timeout: some_huge_number)
You need to remember that LIMIT + OFFSET pagination is prone to race conditions - it may happen that you’ll miss some rows or see rows that you shouldn’t. Database-level cursor gives a consistent view of data.
@michalmuskala yes we definitely reached for Repo.stream as the first option, but in our case the query is processing ~1M records, making HTTP requests for a subset that match some criteria.
The total time could run for hours, so we opted for the pagination approach. The query orders by record insertion time set by db trigger, and no records are deleted, so in our particular use case the race conditions aren’t a problem.
awesome. how is this render in the phoenix view?
Shouldn’t the primary key order by insertion time automatically?
We used this approach in a background job. For a Phoenix view, I’d probably just use Scriviner with query params for page number and size.
Sequential ids would work, but we used uuids for the primary key on that table IIRC.
LIMIT + OFFSET is also linear. When getting the last 100 in the 1 million, you will go through the first 99900. Using an ID or a stream with a timeout of :infinity would be preferred I would say.
thanks. that was useful
Detailed explanation there: http://use-the-index-luke.com/no-offset
Short version, when iterating over the records get the max range and the min range to make sure it’s used in your where clause. This trims the result set BEFORE sorting. Using LIMIT and OFFSET means the entire possible result has to be loaded and sorted to calculate which chunk of them is coming back.
Thanks @josevalim, TIL!
There are some trade-offs with long running transactions also:
If none of these are problematic then I agree Repo.stream with an infinite transaction timeout is a very clean solution.
How about selecting the data into a temporary table with sequential primary key?
You could then use where conditions to read the data in chunks, without being prone to race conditions or inefficient querying.
Your chunk function is very nice. A possible small improvement: return an error if queryable.order_bys is [], since (as you said) pagination will be unpredictable without a consistent sort criteria.
Also, if you just want to know the absolute fastest way to get the data out in the smallest footprint, you’ll want to use Postgres COPY. It’s the fastest way to get data into or out of PG (by a wide margin).
Probably not what you really need in this particular situation, but if you really wanted to you could use it to write the data to a file (like a cache) and then stream over the lines of the file.
Used this approach with a “cron job” problem. There was one very difficult stage in our implementation where it makes an HTTP request using the information retrieved from the db. If for some reason the HTTP call times out and raises an exception, the whole thing is brought down. The stage processes are restarted but it doesn’t recover the “job” midway. The job is basically stopped. It doesn’t even redo it from start. It’s cool when it works though.
Nowadays Scriviner added new option: allow_overflow_page_number, which defaults to false.
This chunk function is relying on empty results to stop streaming data, but with this option empty result is impossible.
To use this chunk function, you need to override this option, like this:
page =
paginate(queryable,
page: page_number,
page_size: chunk_size,
options: [allow_overflow_page_number: true]
)
There is another possibility no one mentioned. You can turn the database table into a work queue itself.
I am currently in a process of doing so on a project, where we initially had the solution @josevalim presented here, i.e. Transaction + Repo.stream + async workers, but we’re ditching it for the reasons of keeping the connection while it’s not necessary to, but also for the reason that this causes race condition if you have 2+ nodes on the cluster. So you have to make sure this process that iterates over database records is initialized on one of the nodes, which requires something like global registration and just complicates things a lot.
So, in your case you could, I think, simply use Honeydew + Ecto queue. https://github.com/koudelka/honeydew I think @koudelka could confirm since he’s on the forum, but that is going to work in your case, and I think pretty well. What it does is adds couple of fields to the table and then does a smart update to lock the next available row, it can also handle retries and failed jobs etc. fairly easily.
We actually used Honeydew on the same project before, replaced it with Exq for the most part, but we are re-introducing Honeydew-like behavior by doing what it does, just slightly differently. We need a bit more flexibility in how and which records we pick, and Honeydew really assumes you want to process all records one by one after they been inserted, and we need something different: processing subset of records at undefined moment in time in future when some conditions change (account subscription status to be precise). So we will not use Honeydew but basically do the same it does in order to lock rows and go one by one through the queue.
For the reference, these are the building blocks you need to build the queue:
https://dba.stackexchange.com/a/69497 - check out the section that talks about “FOR UPDATE SKIP LOCKED” as this is crucial to building queue system on top of PostgreSQL without race conditions and deadlocks, and also how they do it in Honeydew
And if you have a process that goes like that over list of rows from database, fetching and locking one after another - there is no reason why you can’t make this parallel also throughout the cluster. You just start a supervisor that starts say 5 workers on each node on the cluster, and you can have 5* N nodes workers without actually blocking 5 * N database connections.
I will probably put together a blog post once this is running on our end, but that won’t be today 
Yep! This is the sort of scenario that Honeydew’s Ecto Queue was written for, the case where there’s a strictly 1-to-1 relationship between a database row and a job (or a set of different kinds of jobs). As you mentioned, Honeydew will handle failure + (delayed) retry + success bits, as well as manage workers in the cluster.
In @joaquinalcerro’s case, each Honeydew worker could hold a persistent connection to the Gmail API. Alternatively, the workers can check out connections from a pool, if the additional complexity is worth it.
Regarding your use case, have you seen Honeydew’s Execution Criteria? You can provide an SQL fragment to tell the queue to only run jobs when they meet a certain condition. For example, when subscription_status = 'paid'.
Honeydew doesn’t require that jobs be run as soon as their associated row is inserted, they’ll be executed if one hasn’t been run successfully before, and if the row meets the criteria.
Definitely reach out if I can help!
As a workaround for this problem I recommend streaming rows using cursor-based pagination.
Then:
Repo.stream does it - and it can cause your database to explode, given your dataset is large enough)For simplicity we published a small library called ecto_cursor_based_stream that gives you a Repo.cursor_based_stream/2 function that mimics the Repo.stream/2 interface. Feel free to use it 
You are not using database cursors but rather several queries over an indexed increasing field, correct? I was just a bit confused by the name
Yes.
P.S. Didn’t know database cursors are a thing. Curious if anybody’s using that in production. (How easy would to encapsulate that with a library? Would it be better?)
Hence the DB cursors are less known.
© Copyright Elixir Forum | Terms | Privacy & Cookies
