apache / apache/druid

remove unnecessary synchronization overhead from complex Aggregators

Open
#8,031 9 comments 0 reactions 0 assignees View on GitHub
Design Review Proposal
Dominant language
Java
Stars
14.1k
Forks
3.8k
Avg merge
2d 58m
Merged PRs (30d)
233

Description

### Motivation

Many complex [Buffer]Aggregator implementations need to add synchronized access to internal data structures due to single-writer-multiple-reader concurrent usage of those during realtime indexing process where they are concurrently queried in addition to getting updated. However, that synchronization is totally unnecessary everywhere else but we pay its price anyway , for example at historical nodes while querying and in batch indexing tasks etc. Most recently this came up in https://github.com/apache/incubator-datasketches-java/issues/263 .

### Proposed changes
I haven't really done a prototype yet but I "think" these changes should be doable.

Add following methods (with default implementations) to `AggregatorFactory` .
```
public Aggregator factorize(ColumnSelectorFactory metricFactory, boolean isConcurrent)
{
return factorize(metricFactory);
}

public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory, boolean isConcurrent)
{
return factorizeBuffered(metricFactory);
}
```
And, replace all calls inside druid code from `AggregatorFactory.factorize[Buffered](ColumnSelectorFactory)` to `AggregatorFactory.factorize[Buffered](ColumnSelectorFactory, boolean isConcurrent)` with right value for `boolean isConcurrent` specified .
`IncrementalIndex` would be made aware of its concurrency context (by changing existing variable `concurrentEventAdd` to `isConcurrent` and it being correctly specified in all places an `IncrementalIndex` instance is created ) so that it can set right value for `isConcurrent` when calling `factorize[Buffered](..)`
Relevant complex aggregator such as `thetaSketch` can then override newly added methods to add synchronization only for cases where it is really needed.

### Rationale

One other option would be that aggregator implementors get additional contextual information (e.g. the `nodeType` they are running on ) and based on that enable/disable synchronization. However, proposed approach is simpler to use for extension writers and takes away the guessing game.
I also contemplated on adding an enum like
```
enum ConcurrencyContext {
NONE
MULTI_WRITE
SINGLE_WRITE_MULTI_READ
...
..
}
```
and using it instead of `boolean isConcurrent` in newly introduced method arguments, but couldn't see any significant advantages of doing that for now.

### Operational impact
None

### Test plan (optional)
Existing unit/integration tests would cover the changes introduced.

### Future work (optional)
Adjust relevant complex aggregator implementations to take advantage of newly added methods.

Contributor guide

Open the contributing guide

Research direction

Start by reading AggregatorFactory and IncrementalIndex, then trace existing factorize and factorizeBuffered calls to identify their concurrency contexts. Inspect the thetaSketch aggregator as an example of a complex aggregator. Done means the new concurrency-aware methods and call sites are consistent, with existing unit and integration tests passing.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
backend, databases, performance
Issue type
Refactor
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.