Druid: Allow sorting segments on some dims before time

[mistercrunch]

@gianm, this is meant as a placeholder, feel free to alter title and content of this issue

In https://github.com/apache/incubator-druid/issues/6066 we discussed the possibility of specifying the sortOrder on ingestion specs.

Currently, rows in segments are sorted based on [__time] + dimensions. Knowing that Druid's bitmaps use some form of RLE compression, and that LZ4 operates by finding duplication on sliding windows (reminiscent of RLE-compression), allowing for a custom sort order could allow for significantly denser segments. Users may want to take the time to specify a sort order improves perf based on their data and workloads. This may mean sorting based on cardinality (lower dims first), one-to-many hierarchies across dims (country->region), or heavily used and/or commonly predicated column.

A key item would be to remove the assumption that __time is the first item in the ordering to allow for more RLE-friendly ordering. As @gianm mentioned in https://github.com/apache/incubator-druid/issues/6066, many places in the codebase build upon this assumption, and we'd have to make sure that isn't the case anymore as a prerequisite.

[gianm]

I changed the title a bit, since we do allow specifying the sort order for every column except __time (it's the order from dimensionsSpec) but we do not let you put time in the middle of them. So this issue may lead to adding a sortOrder parameter, but I wanted the title to capture the problem we're trying to solve.

For anyone reading, check out #6066 for motivation and what workarounds people use today.

As to removing the assumption that __time is sorted, these are the places I can think of where that exists:

1. The makeCursors function in QueryableIndexStorageAdapter is responsible for creating a sequence of time-granular cursors. This will probably be the toughest thing to adjust, since many query engines depend on the time-ordered and time-granularized nature of these cursors (including the timeseries, topN, groupBy, and timeBoundary engines). Probably for timeseries the best thing to do is call through to the groupBy engine, which can handle its grouping keys being out of order, and is capable of grouping on time. For topN and groupBy I'm not so sure. For groupBy in particular keep in mind #1926; the behavior of "granularity" is not equivalent to the behavior of adding a time-floored dimension.
2. makeCursors again: its time filtering functionality assumes that the time column is sorted.
3. SingleScanTimeDimSelector is used for extractionFns on the __time column, and it also assumes the column is sorted.
4. SingleLongInputCachingExpressionColumnValueSelector is used for expressions on the __time column. It doesn't assume sortedness, but the optimization it does may not make as much sense if the column is sorted.
5. The getMinTime() and getMaxTime() functions in QueryableIndexStorageAdapter assume that they should return the first and last row, respectively.

Some other considerations:

• Given (1) we will probably want to translate some granularity based queries into queries that use the timestamp_floor expression. For example, a timeseries with granularity less than the segment interval should get executed the same way as a timestamp_floor groupBy, and we could make that translation under the hood. Therefore we should look into optimizing timestamp_floor. We should expect at least _some_ localized sortedness in the time column and the function should take advantage of that. Currently, it doesn't.
• If it makes life easier for a first version of this patch, we could consider _not_ supporting "granularity" based queries for topN and groupBy on non-time-sorted segments at first. I believe they are less commonly used than granular timeseries, and could be added in later.

[gianm]

Fwiw, some situations where ability to sort by other-than-time would be expected to be useful:

1. Timeseries data (ironically). Timeseries data is usually modeled as "series" of "points" where each series has a "metric" (like its name) and "tags" that can be used to differentiate it from other series with the same "metric". In Druid you'd model this by making each point into a row, and making the metric and tags into dimensions. It's best to store these rows sorted by metric, so the rows for a particular metric compress better (since they are likely to have a lot of the same tag values), and so we can retrieve all the points for a series faster (since they will have better locality of storage).
2. Clickstream data when you want to do session analyses on it. The idea would be to partition by day first, then by session id (we do already support this: segmentGranularity DAY, and single-dimension partitioning via Hadoop indexing). Then within a segment, sort by session id. It makes it possible to do queries like "count the number of sessions where X, then Y, then Z happened" in linear time and constant memory.
3. Multi-tenant datasets, where you store data for different tenants in the same Druid dataSource. In this case you'd want to both partition and sort by tenant_id. It should improve both compression ratio and query time.

[stale[bot]]

This issue has been marked as stale due to 280 days of inactivity. It will be closed in 2 weeks if no further activity occurs. If this issue is still relevant, please simply write any comment. Even if closed, you can still revive the issue at any time or discuss it on the [email protected] list. Thank you for your contributions.

[stale[bot]]

This issue has been marked as stale due to 280 days of inactivity. It will be closed in 4 weeks if no further activity occurs. If this issue is still relevant, please simply write any comment. Even if closed, you can still revive the issue at any time or discuss it on the [email protected] list. Thank you for your contributions.

[stale[bot]]

This issue has been closed due to lack of activity. If you think that is incorrect, or the issue requires additional review, you can revive the issue at any time.