[Proposal] Vectorized Aggregations

[xvrl]

It seems like vectorization is all the rage these days across the big data world, and a few people have pinged me about the work I had started at some point, but never quite finished.

The gist of the initial work can be found in the following branch (this is just a "no-filter double-sum timeseries", and "multi-value top-n" proof of concept)
https://github.com/druid-io/druid/compare/master...metamx:vectorize-take2?expand=1

The basic idea is to add two new methods on the aggregators:

• a block aggregation method that can aggregate an entire column chunk at once
• a parallel aggregation method that can aggregate multiple buffers at once.

The parallel aggregation can be used to do do multiple dimension values at once in top-n for example.

The block aggregation can massively speed up full column aggregations, thanks to the JVM being able to emit data parallel instructions (e.g. AVX) when you loop unroll the aggregations.

For block aggregation however, it is not enough to just loop-unroll. To achieve significant improvements, we also want to do block reads from the column and avoid materializing the filter offsets if possible.

To that effect we also need to add:

• Methods to block reads to the columns to return arrays of float / long
• Methods to block increment the offsets.

That's where things stand in the current proof of concept, which according to the benchmarks in #2875 give us a 2.6x speedup for a single column double-sum timeseries without filter, and a 1.6x speedup for a double-sum top-n on a multi-value column with 6 values per row.

# Run complete. Total time: 00:02:13

Benchmark                                      (numSegments)  (rowsPerSegment)  (schemaAndQuery)  (vectorize)  Mode  Cnt      Score     Error  Units
TimeseriesBenchmark.querySingleQueryableIndex              1            750000           basic.A         true  avgt   25   9557.090 ± 182.812  us/op
TimeseriesBenchmark.querySingleQueryableIndex              1            750000           basic.A        false  avgt   25  25360.562 ± 597.906  us/op

# Run complete. Total time: 00:02:15

Benchmark                                (numSegments)  (rowsPerSegment)  (schemaAndQuery)  (threshold)  (vectorize)  Mode  Cnt       Score      Error  Units
TopNBenchmark.querySingleQueryableIndex              1            750000           basic.A           10         true  avgt   25   95951.322 ± 1541.184  us/op
TopNBenchmark.querySingleQueryableIndex              1            750000           basic.A           10        false  avgt   25  157396.914 ± 2099.387  us/op

So far those changes only handle the non-filtered aggregation. In order to support efficient aggregation for filtered queries, we also need to manage filtered offsets in a block-like manner.

I talked a little bit more about it with @gianm, and based on our conversation we've come up with the following ideas to tackle this:

We can change ReadableOffsets to be able to return an array of offsets that match for the current block, e.g.

public interface ReadableOffset
{
  /**
   * Returns either an array of offsets for the current row block (if non-contiguous) or an array in which the first
   * element is negative and indicates -first_offset for a contiguous block.
   */
  int[] getOffsets();
}

The offset array can then be pre-allocated by the engine, and then used by the column selector to return a coalesced block of the values that should be aggregated.

Note: these blocks may need to be cached to avoid thrashing of compression buffers when we have multiple selectors for each column.

To simplify things further, we'll probably want to get rid of on-heap aggregators, so we don't have to implentn four different implementations for each aggregator and only have vectorized and non-vectorized at the query engine level (we can keep the on-heap non-vectorized ones for merging)

We'd love to hear people's thoughts, and @gianm please feel free to add anything noteworthy I may have left out from our conversion. Thanks!

[gianm]

I can back this statement up:

For block aggregation however, it is not enough to just loop-unroll. To achieve significant improvements, we also want to do block reads from the column and avoid materializing the filter offsets if possible.

I did test out what performance would be like if we only unrolled aggregator loops (but didn't do anything smart on the column-reading side, just looped over the existing Offset and column-reading interfaces). The speedup on a single-longSum-agg timeseries query was about 20%, which is good but not as good as the 60% that @xvrl got by also doing block reads and avoiding offset materialization.

Since @xvrl's implementation didn't support filters though, it remains to be seen where in between 20% and 60% we would end up with something that does support filters but is also a bit smarter than just looping over the existing interfaces like I did.

[sascha-coenen]

Does this PR relate to Apache Arrow in some way? I believe I read somewhere that you guys had this framework under consideration at some point.

[fjy]

@sascha-coenen No.

We've never really considered Arrow, however the issue that once completed will eventually lead to Druid supporting extendable segment formats including Arrow is here: #2965

[sascha-coenen]

I don't mean to spam your discussion with my input as I'm not too knowledgeable about the Druid internals, but if I may, I'd like to propose something while not knowing whether things already work that way or not nor do I know whether the proposal is sound or not.

To my understanding, segments do currently not contain any aggregates that have a higher granularity than that of the rolled up records.
If a segment's block for a given metric had an aggregate value for the whole block, then one could reduce the computational cost for queries:
If the bitmap of records to aggregate over has less than half of the bits set that make up one block, one would sum these values up as usual. If the bitmap had more than half of the bits set that make up one block, one would aggregate over the inverted bitmap and subtract the result from the pre-aggregated value that is stored for that block.
This would ensure that in the worst case, only half of a block's records would have to be aggregated.
Seeing how the query latency is mostly determined by how many records remain to be aggregated after filtering, it seems that the opportunity of cutting these in half is worth holding the pre-aggregated values.
Where to keep those aggregates is of course another topic. They could go into the segments or be kept in some cache....

[xvrl]

@sascha-coenen it sounds like what you are proposing is some form of materialized views. The question then would be what time granularity to choose for those pre-aggregated records, since that may differ from the underlying segment granularity. Your example case would work for simple sum aggregations, but it would not necessarily extend to complex aggregations or even min / max aggregations, since those do not necessarily support subtracting rows. It sounds like this could help in a small number of use cases, but I'm not sure if it's worth the added complexity. Given that this is unrelated to the topic of this issue, I suggest opening a separate issue if you'd like to discuss further.

[stale]

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 dev@druid.apache.org list. Thank you for your contributions.

[stale]

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.