Scan query: time-ordering

[gianm]

Currently, the Select query is "better" than Scan in one case: it supports time ordering, so you can do queries like "latest 1000 records". But sadly, Select queries are known for excessive memory use (see #5006) so we would like to replace them with Scan whenever possible.

We could support time-ordering for Scan using an approach like the following:

1. Analyze the segment timeline and start from the first (or last, if descending) chunk.
2. If the limit is "low" (below some reasonable number like 100000, let's say) then maintain a priority queue of size = limit, and find the earliest (or latest) rows in the current chunk by scanning each segment in turn. If the current chunk can fill up the priority queue then we are done. If not then move on to the next chunk.
3. If the limit is "high" then that approach won't work: it will use too much memory. Instead, we can do an N-way merge of the individual segments for the current chunk and send those results back to the client.
4. If there are too many segments for an N-way merge sort (100s of segments in the time chunk) then that approach won't work: it will open up too many column selectors at once (each one has overhead: it needs decompression/decoding buffers). Instead, we can do a multi-level merge on disk. This is kind of lame (it will really slow down the query) but it's still better than what Select would do, which is crash the machine.

IMO, implementing only (2) will let people move more workloads to Scan (mostly stuff like "find the most recent X rows"), and so it would be a good start to just do that by itself.

Tagging this as "SQL" too, since if we can improve Scan to handle more cases, then we should also switch over the Druid SQL planner to use it instead of Select in those cases.

[himanshug]

Implementing (3) makes sense to me as well with one additional change that if limit was L then it would work like below.
try finding latest L (or less) in first "hour", say some x < L found (write these x rows to response out)
then try finding latest (L-x) in second "hour"...
... keep going till you either find L rows or have been through all segments.

the "hour" could be "minute", "day" etc ... it is either provided by the user in the query or automatically determined (e.g. use segment granularity). By breaking it down, I'm basically hoping there wouldn't 100s of segments in one "hour" window leading to the problem mentioned in (4).

[KenjiTakahashi]

FYI: I'd just like to see this kept as option in the query, so ones that do no need sorting avoid the performance hit.

I'd also point out that, besides the memory hit, select also tends to be very slow. In some of our cases, it was literally 100's of times vs. scan.

[RestfulBlue]

Current implementation of select query is almost unusable. It cant be limited and can kill entire cluster. i hope this change will be implemented in near future. after that, i think current select query can be removed at all =)

[gianm]

FYI: I'd just like to see this kept as option in the query, so ones that do no need sorting avoid the performance hit.

Agreed, we should keep the default as unsorted.

Current implementation of select query is almost unusable. It cant be limited and can kill entire cluster. i hope this change will be implemented in near future. after that, i think current select query can be removed at all =)

I'd definitely love to be able to remove the select query. I believe its design is not salvageable.

[justinborromeo]

@gianm How would this look from an interface point of view? Would we be adding a new JSON field or is there an existing field where a user could specify to time-order results?

[gianm]

@gianm How would this look from an interface point of view? Would we be adding a new JSON field or is there an existing field where a user could specify to time-order results?

My suggestion would be a new parameter "timeOrder" that is an enum which can be either "ascending", "descending", or "none", with "none" being the default if not provided.

[justinborromeo]

Would it make sense to have both timeOrder and the descending parameter (from BaseQuery)?

[gianm]

Hmm, I don't think descending makes sense for the Scan query. It's a boolean since it makes an assumption that normally queries have an ascending order but that it can be reversed if you want. But that's not really how Scan works, it would have three modes: ascending, descending, and unordered.

IMO, descending should not have been added to BaseQuery in the first place. Not all query types use it. In the GroupBy and Search queries, it's always false. GroupBy does have an ordering concept, but it's more powerful than a simple ascending vs. descending, so it works through the limitSpec instead.

[justinborromeo]

The way Scan is written now, the "scan query doesn't retain all rows in memory before rows can be returned to client" (http://druid.io/docs/latest/querying/scan-query.html) and the design of the code reflects this. Afaik, the proposed implementation for the time-ordering feature would require the broker to store all the returned rows either in memory or on disk before it could start returning rows in a sorted order. It seems that this would violate the main principle behind Scan...is this time-ordering feature something that's appropriate for Scan or am I misunderstanding something?

[gianm]

IMO it’s appropriate and would just need to involve changes to the docs to reflect new behaviors. Scan in that case would be able to start returning rows immediately in unordered mode, and would buffer them for sorting in ordered mode.