sql_select input table pagination?

[scroback77]

Hi,

I've seen the experimental sql_select input (https://www.benthos.dev/docs/components/inputs/sql_select/) and I'm a little unclear on how it works. Does this input do a massive select on a table and read it into memory before sending rows (or batches if specified) to the processors?

Some tables that we read from are massive and have significant performance issues with large selects. We use cockroachdb (which should be supported with the postgres driver), and we are trying to add indexes to our tables so that pagination will be much more efficient. Does the sql_select input paginate on the backend, and if so does it paginate via OFFSET/LIMIT, or with a WHERE on an index? This has a significant impact on whether we can integrate Benthos with our database and processing.

Thanks for the info!

[Jeffail]

Hey @scroback77, the rows returned from the query are streamed so they aren't loaded into memory at the same time by Benthos, but you're still limited by the execution time of the query. The query does support dynamic arguments but pagination is awkward (you'd need to do it by timestamps so it'd be flakey) as you can't reference the context of the latest consumed row.

I'm waiting for feedback on the current state of the component before expanding it but the next logical addition would be the ability to reference the last row consumed in the select statement arguments. We'd need to work out whether we'd need an entire new field to put a separate select for the very first grab, or whether it'd be enough to have a single select statement where the SQL itself would need to cater for the first query where no existing rows are loaded yet.

[webfrank]

Hi, I was trying to move a huge postgres table from one db to another but although the sql_select is streaming it crashes the server.
I would like to achieve a paginated select (with limit and offset). I tried without success using while: but don't know where to store the offset and how to increment it. A component like sql_select with such feature would be very useful

[mihaitodor]

Would a cache help you? https://www.benthos.dev/docs/components/caches/about

[carrascoacd]

Hi, I'm interested in this feature. Looking at this I got into a possible example with the twitter_search template. The thing is that I think that template is not 100% right, but maybe I'm wrong.

If the output used in a pipeline with the example of the twitter_search input fails, the cache will hold the wrong cursor so it will try to fetch new data while the previous one has not been processed yet. Example:

• Input fetches 10 objects. Cache stores loop=1.
• Output fails, so no objects have been processed.
• Input fetches 10 objects. Cache stores loop=2.
• Output fails, so no objects have been processed.
• Input fetches 10 objects. Cache stores loop=3.
• ... and so on

I think the cache should be updated in the last step of the pipeline, just after every object has been processed.

Please, let me know if that input can be used as an example for SQL pagination or you have a better alternative. At least-once is needed in my case.

---

Note: Remember that LIMIT+OFFSET could potentially do a full scan of your table, so try to paginate in another way for large datasets, like using indexes in the WHERE.

[mihaitodor]

Hey @carrascoacd, I'm not super-familiar with that specific input, but all inputs in Benthos need to get acks / nacks from the configured output(s) before those messages are discarded. If the output returns an error, Benthos will keep trying to emit those messages until it succeeds. The max_in_flight field which is present on most outputs along with the batching config control how many messages can be in-flight at the same time in Benthos before the input gets throttled. Some outputs are batched, so, for those max_in_flight reflects the number of parallel message batches that can be in flight at the same time. One of the core goals for Benthos is to ensure at-least-once delivery semantics (see this video for details).

I think the cache should be updated in the last step of the pipeline, just after every object has been processed.

Can you provide more details (and maybe a simplified example) of what issue you're seeing with the cursor that's stored in the cache for this twitter_search input template? This seems to be the last step of the pipeline (except for the log processor), but maybe I'm missing something...

Please, let me know if that input can be used as an example for SQL pagination or you have a better alternative. At least-once is needed in my case.

I have a somewhat complex config here that you can use as a reference. It does SQL pagination with LIMIT and OFFSET, but I needed to have it query pages in parallel, so there's some bloblang logic in there to compute the page bounds after each batch of pages finishes processing. It hasn't been tested exhaustively, so it might still have some issues... Also, if such a complex solution is required, I think it might be better to leverage an external job scheduler such as Airflow.

If you don't need to select pages in parallel, then the config should be simpler and I was meaning to post an example but I didn't get a chance to finish it. I'll get back to it eventually...

LE: @Jeffail mentioned that the "twitter_search input has always stored latest received messages on consume rather than on ack, its a limitation of the fact that its a template" & "if we wanted to productionise that input wed need a proper implementation". If you need it quickly, one way to do it is via a custom plugin: https://github.com/benthosdev/benthos-plugin-example

[carrascoacd]

LE: @Jeffail mentioned that the "twitter_search input has always stored latest received messages on consume rather than on ack, its a limitation of the fact that its a template" & "if we wanted to productionise that input wed need a proper implementation". If you need it quickly, one way to do it is via a custom plugin: https://github.com/benthosdev/benthos-plugin-example

Thanks @mihaitodor for your response. I didn't know about that mention, it is everything clear to me.

I have a somewhat complex config here that you can use as a reference.

In this example, I think you have the same problem, if the sink fails and you get a restart the pipeline already stored the "next batch", the state of that specific moment is lost. I think that doing the ack to the cache (set) in the output (or in your case after the page_selector run) will ensure consistency since you only "acknowledge" (ie set the new value into the cache) after it has been processed. The idea is basically acknowledging every loop at the very end, so we always know that our batch is processed at least once.

[mihaitodor]

@carrascoacd Yep, you're right. I don't know what would be a good way to address this limitation in templates... Maybe if the template uses a cache to do pagination, then it should also take as an extra parameter a list of processors so the user can inject their custom processors that way?

[carrascoacd]

I think that the best choice would be to allow messages that were processed in the last loop to be accessible from the input, then with all the choices we can hit a cache, interpolate or do whatever. Maybe is not a limitation of the template but from the input that requires complex association of different inputs to loop and ack messages (in this case, rows).

[mihaitodor]

Dunno... I think that's something for @Jeffail to consider