Theta sketch - Concurrent union implementation

[Eshcar]

The implementation of a concurrent theta sketch is based on two main design choices

1. updating threads write into a local buffer which is propagated in the background to a shared sketch
2. query threads read from a snapshot created by the shared sketch, namely they always see a consistent state of the shared sketch.
   More details can be found here https://datasketches.github.io/docs/Theta/ConcurrentThetaSketch.html

This issue discusses 3 design alternatives for implementing a concurrent union operation.

Design option I: This design works very similar to the design of concurrent theta sketch. It allows updating only the local buffers, and querying only the shared union object.

1. ConcurrentSharedUnionImpl extends Union overrides all update methods with UnsupportedException
2. ConcurrentHeapUnionBuffer extends Union overrides getResult and getByteArray methods with UnsupportedException
3. Add to SetOperationBuilder 2 methods:
   -buildShared gets ConcurrentSharedThetaSketch returns ConcurrentSharedUnionImpl
   -buildLocal gets ConcurrentSharedUnionImpl returns ConcurrentHeapUnionBuffer

Design option II: This design has some pros over the first design alternative mainly simplicity, however it means the “backend” is a shared sketch, and querying the union object is done through one of the local buffers, which is only available to the updating threads. Question: is this a reasonable setting?

1. ConcurrentHeapUnionBuffer extends Union supports all methods both updates and queries
2. Add to SetOperationBuilder 1 method:
   -buildLocal gets ConcurrentSharedThetaSketch returns ConcurrentHeapUnionBuffer

Design option III: This design is very similar to the second option however users cannot query the local buffer. This design assumes that queries can be delegated to the ConcurrentSharedThetaSketch “backend” which is available somewhere in the system. Question: is this a reasonable setting?

1. ConcurrentHeapUnionBuffer extends Union overrides getResult and getByteArray methods with UnsupportedException
2. Add to SetOperationBuilder 1 method:
   -buildLocal gets ConcurrentSharedThetaSketch returns ConcurrentHeapUnionBuffer

[Eshcar]

per @leventov suggestion forwarding the questions in this issue to the Druid dev list

[himanshug]

updating threads write into a local buffer which is propagated in the background to a shared sketch

that sounds like there are extra threads doing things in the background. I wonder if that and needs for creating snapshot offsets(or worsens) the performance improvements received by removing the lock in the context of Druid. As bulk of the use case in Druid is single threaded access except for in-memory data on real time indexers as discussed in detail on https://lists.apache.org/thread.html/9899aa790a7eb561ab66f47b35c8f66ffe695432719251351339521a@%3Cdev.druid.apache.org%3E where the access is single-writer-multiple-reader

[gianm]

Speaking as a Druid person & adding to @himanshug's comment - there are two main uses for aggregators (whether on- or off-heap) in Druid right now:

1. At indexing time we use aggregators for doing rollup (summarization). Currently this is the on-heap aggregator flavor, but there are some discussions about moving it to the off-heap flavor. This is the single-writer multiple-reader case @himanshug was talking about.
2. At query time we use aggregators for doing group-by, topn, timeseries, etc queries. Depending on the query engine it might be the on-heap or off-heap flavor. Access here is never concurrent, so any overhead to support concurrency would be needless overhead in this case.

For most users probably the bulk of CPU time is spend on (2)

[Eshcar]

Thanks @himanshug, @gianm,

I am aware of the work done by Anastasia and Eran on integrating Oak into Druid, and follow the Issues and PRs very closely :)

Let me elaborate some more on our implementation and benchmarks results.

Ingestion time aggregators: single-writer-multiple-readers
From the Oak-related discussions and also from our previous conversation on the Druid dev list I understand the real-time indexer access is single-writer-multiple-reader.
Concurrent Union (and in general concurrent sketch) is also beneficial for this use case, specifically when handling big aggregations.
We've run benchmarks that demonstrated the benefits of single thread concurrent theta sketch over lock-based implementation, when sketches are big - I can share the results, or better I should publish them in the data sketch web site.
We mainly measured write throughput, since it is hard to measure read latency percentiles with high accuracy. It is likely to assume that the latency of readers improve as they are simply reading an attribute instead of taking a lock and accessing more than one attribute.
I believe we will reproduce these results for concurrent Union.

that sounds like there are extra threads doing things in the background. I wonder if that and needs for creating snapshot offsets(or worsens) the performance improvements received by removing the lock in the context of Druid.

You are right there is a background thread. For small sketches we have a workaround where the updating thread itself does the propagation. Creating a snapshot for theta sketch is very simple and efficient. When considering single writer thread in small sketches indeed the overhead of managing local buffers and shared sketch offsets the benefit of removing the lock, but from a certain size (number of uniques) the improvement outweighs the overhead.

Query time aggregators
This is the case where there are no updates concurrent to queries (right?). So the current implementation does not use locks. Our evaluation results show that when the sketches are VERY big (more than 1M uniques) using multiple threads can double and even triple the updating rate (vs single thread with no lock). So if there is a way to identify potential heavy aggregation queries, considering running multiple threads of the concurrent implementation can result in a significant reduction of query latency.
We never tested sketches in the context of Druid. Once we have the implementation of the concurrent Union we can start playing with Druid queries.

Back to my original question, I think I will implement design option 1, it means implementing one additional class, and have many method throwing exception, but it is the most reasonable option in order to be compatible with application which use the existing union implementation.

Long answer :)
Please let me know if there are more comments or questions.
And thanks again for the feedback.

[himanshug]

When considering single writer thread in small sketches indeed the overhead of managing local buffers and shared sketch offsets the benefit of removing the lock, but from a certain size (number of uniques) the improvement outweighs the overhead.

from what I have seen, sketch size typically follows the power law with large sketches being at the tail end.
in any case, I would be slightly worried about background threads getting introduced that require extra thinking about who spawns them, how many are spawned, are they daemon threads, who kills them etc .

also, apache/druid#8031 will hopefully remove synchronization usage at non-realtime data query(and other single thread access places such as batch indexing) altogether for all complex aggregator implementations in Druid.

[Eshcar]

Background threads are daemon threads that can be managed in a limited size pool.
I can imagine large sketches are at the tail end of a power law distribution.
But if in some scenarios these large sketches are responsible for tail latency (both in ingestion and in queries) then this could be a good motivation for giving them special care.

Are you aware of high latency queries due to long merge/union operations that can be accelerated with parallelism?

BTW - in the mixed workload benchmarks I ran for concurrent theta sketches, I only used read threads as "background" noise, with some sleeping time every few queries.
Do you have an estimate on the rate of queries against real-time index in Druid?

[himanshug]

unfortunately it is very difficult to have answers for "expected query rate at real-time index" etc because it heavily depends on cluster to cluster.
that said, even on realtime indexing process, IncrementalIndex (only place with concurrency) is persisted to a disk based structure periodically(or hitting memory/row thresholds) because disk based structure is highly optimized for read/queries (not just aggregations but it has other things like bitmap indexes for filtering, takes relatively less memory backed by disk so not counted as "anonymous memory" on linux) . "on most clusters", I wouldn't expect a large number or large sketch queries happening on IncrementalIndex .

[Eshcar]

I get the message. Locks are a good enough solution for incremental index since 1) ingestion is done by a single thread, and vast majority of sketches are small to medium size, 2) reads arrive at a low rate, and queries rarely need to read or aggregate large sketches.
Fair enough :)

A question that is not related to incremental indexing.
I assume aggregation at query time are executed in memory, even when data already resides on disk.
What I suggest here is to consider a way to expedite query processing by having multiple threads (can be a small number) executing the aggregation in parallel, instead of sequentially by a single thread.
For example if a query needs to aggregate (union) thousands of sketches then perhaps parallel processing is better.
I believe this can reduce query latency, and specifically significantly reduce high percentiles query latency. It should be straightforward to understand if a query needs to aggregate a large number of sketches and only for these cases trigger parallel union (by using concurrent union).

Does this seems like something that can improve Druid performance?

[himanshug]

assume aggregation at query time are executed in memory, even when data already resides on disk.
What I suggest here is to consider a way to expedite query processing by having multiple threads (can be a small number) executing the aggregation in parallel, instead of sequentially by a single thread.

At one index/aggregator level , it might look like that we are not exploiting all available cores and adding concurrency might help. However, in reality there are many queries running concurrently, each individual query is further split into per segment scan/aggregation job that runs concurrently. That does exhaust available cores.

[Eshcar]

Do you have a standard way to stress test the system by running many queries agains it while measuring their performance (50th, 95th and 99th query latency percentiles) -- in case I would like to give it a try.
What would you recommend - a standalone machine cluster is sufficient for stress test, or a 3 machine cluster is more representative for real world clusters?

And thanks a lot for your patient in explaining all the above.

[himanshug]

usually testing happens on user clusters. You could probably use one of the publicly available datasets like http://www.tpc.org/tpch/ . In this context 1 broker + 2 historical + other_nodes_on_any_number_of_machines should be enough . that wouldn't necessarily represent a real world cluster but good enough for this experiment I guess.

[Eshcar]

Thanks.
Actually we (Verizon Media) have enough data sets that are stored in Druid, so that should be ok.
After I implement the new Union I will set up a small cluster and will try to experiment with the new code. It might take a while since I am not 100% on this project.
Once I have results I will ping you again.

[leerho]

@Eshcar Are you still working on this?

1. Are there code changes that need to be made to the current code base?
2. If and when can we close this issue?

[leerho]

Closing this issue due to lack of response. We can reopen this issue if necessary.