Add result level caching to Brokers #5028

a2l007 · 2017-10-31T20:40:10Z

Based on proposal #4843 , this introduces result level caching in brokers. It uses the etag functionality to identify the existence of result cache for a specific query. This is independent of the segment level caching and therefore both types of caching can be configured independently. A new query runner: ResultLevelCachingQueryRunner is introduced that performs this caching. The result level cache is populated after the merge using the query key as the cache key and merged result is saved as the cache value along with the etag information.
Since query results may be large, there is a configurable parameter resultLevelCacheLimit that limits the size of the query response that can be cached.
One caveat to be noted is that basic object deserialization is performed while retrieving post aggregated values from the cache. Therefore this may not work with post aggregators that require a custom deserialization method.

drcrallen · 2017-10-31T20:45:12Z

processing/src/main/java/io/druid/query/QueryContexts.java

@@ -37,6 +37,8 @@
  public static final boolean DEFAULT_BY_SEGMENT = false;
  public static final boolean DEFAULT_POPULATE_CACHE = true;
  public static final boolean DEFAULT_USE_CACHE = true;
+  public static final boolean DEFAULT_POPULATE_RESULTLEVEL_CACHE = true;


New feature should start off disabled

Would it be sufficient to keep it disabled via the populateResultCache parameter?
ba78816#diff-661a9dbd25633c651845a9ecab0ddbaaR44

It is confusing to have default here be true but the default in CacheConfig be false. But this just follows the convention of the prior caching settings.

Can you add a comment in CacheConfig that the defaults as stated in QueryContexts are different due to legacy reasons, and should probably be made the same at some point in the future?

Added comment as suggested.

drcrallen · 2017-10-31T20:45:40Z

cool! Any chance you can get empirical evidence on the hit rates for this feature?

leventov · 2017-10-31T21:06:52Z

server/src/main/java/io/druid/client/CachingClusteredClient.java

+      }
+      final Function<Object, T> pullFromCacheFunction = strategy.pullFromCache();
+      final TypeReference<Object> cacheObjectClazz = strategy.getCacheObjectClazz();
+      Sequence<Object> cachedSequence = new BaseSequence<>(


It could be Sequences.simple(() -> {...lambda, returning iterator...})

leventov · 2017-10-31T21:07:54Z

server/src/main/java/io/druid/client/CachingClusteredClient.java

        List<Sequence<T>> sequencesByInterval = new ArrayList<>(alreadyCachedResults.size() + segmentsByServer.size());
        addSequencesFromCache(sequencesByInterval, alreadyCachedResults);
        addSequencesFromServer(sequencesByInterval, segmentsByServer);
        return Sequences
            .simple(sequencesByInterval)
            .flatMerge(seq -> seq, query.getResultOrdering());
+      }).map(r -> {


Please refactor as a separate method

drcrallen · 2017-11-03T16:53:57Z

server/src/main/java/io/druid/client/CacheUtil.java

+  )
+  {
+    return new Cache.NamedKey(
+        resultLevelCacheIdentifier, resultLevelCacheIdentifier.getBytes()


StringUtils.toUtf8 instead of getBytes is suggested

drcrallen · 2017-11-03T16:59:52Z

server/src/main/java/io/druid/client/CachingClusteredClient.java

+      @Nullable
+      final byte[] queryCacheKey = computeQueryCacheKey();
+      @Nullable
+      final String queryResultKey = computeCurrentEtag(segments, queryCacheKey);


This now computes the etag for every query regardless of if caching is enabled.

Can this only be computed if it will actually be used?

drcrallen · 2017-11-03T17:04:18Z

server/src/main/java/io/druid/client/CachingClusteredClient.java

+      @Nullable
+      final byte[] cachedResultSet = fetchFromResultLevelCache(queryResultKey);
+      if (cachedResultSet != null) {
+        log.info("Fetching entire result set from cache");


This should be debug, not info. That's a lot of logging to have on every query.

drcrallen · 2017-11-03T17:04:52Z

server/src/main/java/io/druid/client/CachingClusteredClient.java

        if (currentEtag != null && currentEtag.equals(prevEtag)) {
          return Sequences.empty();
        }
      }

+      @Nullable
+      final byte[] cachedResultSet = fetchFromResultLevelCache(queryResultKey);


This should only happen if the using the results cache is enabled.

drcrallen · 2017-11-03T17:08:47Z

server/src/main/java/io/druid/client/CachingClusteredClient.java

      final List<Pair<Interval, byte[]>> alreadyCachedResults = pruneSegmentsWithCachedResults(queryCacheKey, segments);
      final SortedMap<DruidServer, List<SegmentDescriptor>> segmentsByServer = groupSegmentsByServer(segments);
-      return new LazySequence<>(() -> {
+      return Sequences.wrap(new LazySequence<>(() -> {


How does this impact the lazy evaluation? Where is the cache work done? in the jetty thread?

Lazy evaluation would continue to work as it is. The wrapped after() method would run after all the results have been accumulated and populates the cache in the jetty thread.

drcrallen · 2017-11-03T17:09:22Z

server/src/main/java/io/druid/client/CachingClusteredClient.java

      });
    }

+    private T cacheResultEntry(T result, String queryResultKey)


as per the other methods, this should return null immediately if queryResultKey is null

The resultcachepopulator entry will be created only if the result cache is enabled and this method will add future entries only if the populator entry is present. This method would be harmless if resultcache is disabled.

drcrallen · 2017-11-03T17:10:57Z

server/src/main/java/io/druid/client/CachingClusteredClient.java

+          );
+        }
+        catch (IOException e) {
+          throw new RuntimeException(e);


Failing to fetch or parse from cache should NOT cause a complete failure of the query. It should continue on and do what it can to compute things.

I'm not completely clear where this will be evaluated, and if the exception is caught and ignored or not.

drcrallen · 2017-11-03T17:12:41Z

server/src/main/java/io/druid/client/CacheUtil.java

@@ -66,7 +75,9 @@ public static void populate(Cache cache, ObjectMapper mapper, Cache.NamedKey key
          gen.writeObject(result);
        }
      }
-
+      if (cacheLimit != 0 && bytes.size() > cacheLimit) {


can this be cacheLimit > 0 and pass in -1 when you don't want to do it?

drcrallen · 2017-11-03T17:13:18Z

server/src/main/java/io/druid/client/CachingQueryRunner.java

@@ -181,7 +181,7 @@ public void run()
            public void run()
            {
              try {
-                CacheUtil.populate(cache, mapper, key, Futures.allAsList(cacheFutures).get());
+                CacheUtil.populate(cache, mapper, key, Futures.allAsList(cacheFutures).get(), 0);


as per above, suggest making this -1

drcrallen · 2017-11-03T17:14:31Z

server/src/main/java/io/druid/client/cache/CacheConfig.java

@@ -46,6 +52,9 @@
  private int cacheBulkMergeLimit = Integer.MAX_VALUE;

  @JsonProperty
+  private int resultLevelCacheLimit = 10485760;


why not Integer.MAX_VALUE by default?

drcrallen · 2017-11-03T17:14:54Z

server/src/test/java/io/druid/client/CachingQueryRunnerTest.java

@@ -351,7 +351,8 @@ private void testUseCache(
        cache,
        objectMapper,
        cacheKey,
-        Iterables.transform(expectedResults, cacheStrategy.prepareForCache())
+        Iterables.transform(expectedResults, cacheStrategy.prepareForCache()),


as per above, suggest -1

drcrallen · 2017-11-03T17:16:29Z

This is a neat feature but I don't know how much battle testing the etag computation code paths have. Some of the underlying code looks racy in ways that can cause hash computation failures. As such, since this is also a new feature, I suggest making sure the code paths will not touch any new code (or will immediately return null) unless this feature is enabled.

drcrallen · 2017-11-03T17:19:14Z

The racy part I'm not sure about is that the etag computation uses ServerSelector's pick() method, which may have an underlying change in discovery state after the initial server selection. But fixing or changing that is beyond the scope of this PR.

drcrallen · 2017-11-03T17:22:03Z

Also please make sure there's a coherent story on what thread is doing computation. It is a notoriously tricky thing to track, and cache populating is actually a very compute intensive operation.

a2l007 · 2017-11-03T18:38:13Z

@drcrallen I'm currently testing the changes on one of our internal clusters and should be able to give you a better response once it is done. I have marked this PR in progress for now.

…t_level_cache

himanshug

I think we should do result level caching business at the very end of query execution by adding a ResultLevelCachingQueryRunner that is added at very top of the query runner chain as in #4852 for setting up context.
Major reason is to have clean separation from existing classes. Also I'm not sure whether current implementation caches raw sketches or finalized values ... it should definitely cache finalized values if not already.

himanshug · 2017-11-15T20:54:25Z

docs/content/configuration/broker.md

@@ -104,6 +104,9 @@ You can optionally only configure caching to be enabled on the broker by setting
 |--------|---------------|-----------|-------|
 |`druid.broker.cache.useCache`|true, false|Enable the cache on the broker.|false|
 |`druid.broker.cache.populateCache`|true, false|Populate the cache on the broker.|false|
+|`druid.broker.cache.useResultLevelCache`|true, false|Enable result level caching on the broker.|false|
+|`druid.broker.cache.populateResultLevelCache`|true, false|Populate the result level cache on the broker.|false|
+|`druid.broker.cache.resultLevelCacheLimit`|positive integer or 0|Maximum size of query response that can be cached.|`Integer.MAX_VALUE`|


what would be the meaning of setting it 0 ? that would disable the cache.

himanshug · 2017-11-15T21:00:35Z

server/src/main/java/io/druid/client/CachingClusteredClient.java

@@ -375,6 +423,7 @@ private String computeCurrentEtag(final Set<ServerToSegment> segments, @Nullable
      Hasher hasher = Hashing.sha1().newHasher();
      boolean hasOnlyHistoricalSegments = true;
      for (ServerToSegment p : segments) {
+        log.info(p.getServer().pick().getServer().getType().toString());


this looks like left here by accident.

himanshug · 2017-11-15T21:18:01Z

@drcrallen current etag feature is being used in some druid clusters at Oath and some users have result level caches built using the etag feature outside of Druid already. Result level Cache hit ratio typically varies from 50% to 80% depending upon the use case.
Another major advantage of result level cache is that you cache finalized value of sketches rather than sketches themselves which take alot of space if stored inside cache.

Regarding thread used for computation, I think we should ensure that cache population activity happens inside the thread that calls ResultLevelCachingQueryRunner.run(..) (I think its gonna be the jetty thread) which should be introduced instead of changing CachingClientQueryRunner too much.

…t_level_cache

himanshug · 2017-11-29T00:29:08Z

server/src/main/java/io/druid/client/CachingClusteredClient.java

+        if (newResultCacheKeyFromEtag != null && newResultCacheKeyFromEtag.equals(prevResultCacheKeyFromEtag)) {
+          return Sequences.empty();
+        }
+      }


this kind of short circuiting is already done by checking QueryResource.HEADER_IF_NONE_MATCH in the query context can we reuse same and not add this block ?

Fixed as per suggestion.

…t_level_cache

a2l007 · 2018-02-15T19:15:55Z

@drcrallen could you please review

drcrallen · 2018-02-16T23:54:03Z

Yes, I'll put it on my list.

drcrallen · 2018-03-09T16:38:10Z

docs/content/querying/query-context.md

@@ -15,6 +15,8 @@ The query context is used for various query configuration parameters. The follow
 |queryId          | auto-generated                         | Unique identifier given to this query. If a query ID is set or known, this can be used to cancel the query |
 |useCache         | `true`                                 | Flag indicating whether to leverage the query cache for this query. When set to false, it disables reading from the query cache for this query. When set to true, Druid uses druid.broker.cache.useCache or druid.historical.cache.useCache to determine whether or not to read from the query cache |
 |populateCache    | `true`                                 | Flag indicating whether to save the results of the query to the query cache. Primarily used for debugging. When set to false, it disables saving the results of this query to the query cache. When set to true, Druid uses druid.broker.cache.populateCache or druid.historical.cache.populateCache to determine whether or not to save the results of this query to the query cache |
+|useResultLevelCache         | `true`                                 | Flag indicating whether to leverage the result level cache for this query. When set to false, it disables reading from the query cache for this query. When set to true, Druid uses druid.broker.cache.useResultLevelCache to determine whether or not to read from the query cache |


as per below, I thought this defaulted to false?

Fixed the docs.

…t_level_cache

drcrallen · 2018-03-19T17:54:18Z

server/src/main/java/io/druid/query/ResultLevelCachingQueryRunner.java

+                "ResultLevelCachePopulator cannot be null during cache population"
+            );
+            if (thrown != null) {
+              log.error(


(minor) why not put the error as a parameter to error(?

drcrallen · 2018-03-19T17:55:27Z

server/src/main/java/io/druid/query/ResultLevelCachingQueryRunner.java

+        }
+      }
+      catch (IOException ex) {
+        log.error("Failed to retrieve entry to be cached. Result Level caching will not be performed!");


should probably log error here

aka, include the exception in the error parameters for the logger call

drcrallen

This is looking good! the main thing i would like to see addressed is if there's a way to make the result level caching call fit in with the fluent query runner. The switch from fluent style to delegation style is kind of jarring.

drcrallen · 2018-03-19T17:56:36Z

server/src/main/java/io/druid/server/ClientQuerySegmentWalker.java

-                    baseClientRunner,
-                    retryConfig,
-                    objectMapper
+    return new ResultLevelCachingQueryRunner<>(


this feels like it violates the fluent workflow, is there a way to make this work inline with the fluent query runner?

drcrallen · 2018-03-19T17:57:37Z

processing/src/main/java/io/druid/query/CacheStrategy.java

   * @return A function that does the inverse of the operation that the function prepareForCache returns
   */
-  Function<CacheType, T> pullFromCache();
+  Function<CacheType, T> pullFromCache(boolean isResultLevelCache);


Since a lot of calls default to false, would it make sense to add another method that just calls pullFromCache(false) and prepareForCache(false) ? and preserve backwards compat?

…t_level_cache

a2l007 · 2018-03-20T18:50:46Z

server/src/main/java/io/druid/server/ClientQuerySegmentWalker.java

@@ -91,37 +91,42 @@ public ClientQuerySegmentWalker(
  private <T> QueryRunner<T> makeRunner(Query<T> query, QueryRunner<T> baseClientRunner)
  {
    QueryToolChest<T, Query<T>> toolChest = warehouse.getToolChest(query);
+    return new ResultLevelCachingQueryRunner<>(makeRunner(query, baseClientRunner, toolChest),


@drcrallen Does this look ok? I have refactored it a bit but it doesn't exactly follow the fluent style.
Incorporating ResultLevelCachingQueryRunner inside the FluentQueryRunnerBuilder is tricky because the query runner logic needs to keep track of cache value data before and after the baseRunner.run is invoked. Further this query runner is not accessible inside the fluent query runner builder, which makes it a bit more complex.
I have addressed your other comments as well.

io.druid.query.FluentQueryRunnerBuilder.FluentQueryRunner#emitCPUTimeMetric use in the fluent query runner builder has a similar need. If you make a method in the fluent builder that takes the missing parameters (query, objectmapper, cache, cacheconfig) does it work?

@drcrallen Tried that but FluentQueryRunnerBuilder which resides in druid-processing does not have access to ResultLevelCachingQueryRunner as it (along with most of the caching logic) resides in druid-server package.
I attempted to refactor the caching logic into druid-processing, but there are several other dependencies that may have to be moved. I could work on a follow up improvement PR to investigate and perform the refactoring and hopefully make the makeRunner method cleaner.

That must be quite frustrating >.<

Can you clarify this in an issue, and put a link to the issue in a comment above this code.

That way anyone coming in later and wondering why it is like this can have a clear logic path for what needs to change to fix things up.

@drcrallen I've made the requested changes.

…t_level_cache

drcrallen · 2018-03-24T02:11:46Z

Sweet! @a2l007 thanks for sticking this out.

leventov · 2018-10-27T05:31:34Z

processing/src/main/java/io/druid/query/groupby/GroupByQueryQueryToolChest.java

@@ -426,6 +427,11 @@ public Object apply(Row input)
              for (AggregatorFactory agg : aggs) {


@a2l007

final List<Object> retVal = Lists.newArrayListWithCapacity(1 + dims.size() + aggs.size());

a few lines above is not updated wrt getPostAggregatorSpecs. Similar problems in other classes, updated in this PR. Also I noticed that somewhere this sizing was buggy before this PR already.

Atul Mohan added 2 commits October 31, 2017 15:28

Add result level caching to Brokers

ba78816

Minor doc changes

7cb33cd

drcrallen reviewed Oct 31, 2017

View reviewed changes

leventov reviewed Oct 31, 2017

View reviewed changes

leventov added the Area - Cache label Oct 31, 2017

Simplify sequences

83ee76e

leventov added the Design Review label Nov 1, 2017

drcrallen reviewed Nov 3, 2017

View reviewed changes

drcrallen requested changes Nov 3, 2017

View reviewed changes

a2l007 changed the title ~~Add result level caching to Brokers~~ [WIP] Add result level caching to Brokers Nov 3, 2017

drcrallen added the WIP label Nov 3, 2017

Atul Mohan added 5 commits November 10, 2017 10:25

Move etag execution

24a7595

Merge branch 'master' of https://github.com/druid-io/druid into resul…

d2861b9

…t_level_cache

Modify cacheLimit criteria

130ee63

Fix incorrect etag computation

efeb2b2

Fix docs

08b4feb

himanshug requested changes Nov 15, 2017

View reviewed changes

Atul Mohan added 5 commits November 18, 2017 11:59

Merge branch 'master' of https://github.com/druid-io/druid into resul…

3639c0a

…t_level_cache

Merge branch 'master' of https://github.com/druid-io/druid into resul…

e1d9175

…t_level_cache

Add separate query runner for result level caching

c805b92

Update docs

d81d81c

Merge branch 'master' of https://github.com/druid-io/druid into resul…

d738fbc

…t_level_cache

himanshug reviewed Nov 29, 2017

View reviewed changes

Merge branch 'master' of https://github.com/druid-io/druid into resul…

7e3492c

…t_level_cache

Atul Mohan added 3 commits February 2, 2018 15:47

Release bytestream after use

bcddabf

Address PR comments

6c42e2c

Discard resultcache stream after use

d51c21e

himanshug approved these changes Feb 5, 2018

View reviewed changes

drcrallen reviewed Mar 9, 2018

View reviewed changes

Atul Mohan added 2 commits March 9, 2018 13:21

Merge branch 'master' of https://github.com/druid-io/druid into resul…

b8547bb

…t_level_cache

Fix docs

96bbf23

drcrallen reviewed Mar 19, 2018

View reviewed changes

drcrallen requested changes Mar 19, 2018

View reviewed changes

Atul Mohan added 3 commits March 19, 2018 14:38

Merge branch 'master' of https://github.com/druid-io/druid into resul…

28a7a0e

…t_level_cache

Address comments

9dec1cd

Merge branch 'master' of https://github.com/druid-io/druid into resul…

b5b7992

…t_level_cache

a2l007 commented Mar 20, 2018

View reviewed changes

a2l007 mentioned this pull request Mar 22, 2018

Refactor ClientQuerySegmentWalker to fix fluent workflow violation #5517

Closed

Atul Mohan added 2 commits March 22, 2018 09:53

Merge branch 'master' of https://github.com/druid-io/druid into resul…

edd79c0

…t_level_cache

Add comment about fluent workflow issue

7f2d48d

drcrallen approved these changes Mar 24, 2018

View reviewed changes

drcrallen merged commit ec17a44 into apache:master Mar 24, 2018

drcrallen added this to the 0.13.0 milestone Mar 24, 2018

drcrallen added Feature Release Notes and removed Design Review labels Mar 24, 2018

dclim mentioned this pull request Oct 10, 2018

Druid 0.13.0-incubating release notes #6442

Closed

swapnilpandit mentioned this pull request Oct 17, 2018

Exception during sketch aggregations while using Result level cache #6483

Closed

leventov reviewed Oct 27, 2018

View reviewed changes

JackyYangPassion mentioned this pull request Jul 3, 2019

0.12.1 result level cache #8018

Closed

		@@ -426,6 +427,11 @@ public Object apply(Row input)
		for (AggregatorFactory agg : aggs) {

Add result level caching to Brokers #5028

Add result level caching to Brokers #5028

Conversation

a2l007 commented Oct 31, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

drcrallen commented Oct 31, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

drcrallen commented Nov 3, 2017

drcrallen commented Nov 3, 2017 • edited Loading

drcrallen commented Nov 3, 2017

a2l007 commented Nov 3, 2017

himanshug left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

himanshug commented Nov 15, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

a2l007 commented Feb 15, 2018

drcrallen commented Feb 16, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

drcrallen left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

drcrallen Mar 22, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

drcrallen commented Mar 24, 2018

Choose a reason for hiding this comment

a2l007 commented Oct 31, 2017 •

edited

Loading

drcrallen commented Nov 3, 2017 •

edited

Loading

himanshug commented Nov 15, 2017 •

edited

Loading

drcrallen Mar 22, 2018 •

edited

Loading