jtuglu1 commented on code in PR #20815:
URL: https://github.com/apache/druid/pull/20815#discussion_r4504247537
##########
docs/configuration/index.md:
##########
@@ -1967,6 +1968,7 @@ You can optionally only configure caching to be enabled
on the Broker by setting
|`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.includeDatasourceAndIntervalsInResultLevelCacheKey`|true,
false|Include the query's datasource and intervals in the result level cache
key. By default the key has neither, so queries that differ only by interval or
datasource share one cache entry and overwrite each other. Enabling this gives
each of them its own entry, which helps when the same queries repeat over a few
different intervals. Trade-offs: the cache holds more entries, and a query
whose interval changes on every run, such as a rolling time window, adds a new
entry each run that is never read again. Queries whose intervals cover the same
data but differ elsewhere, such as an end time past the latest data, no longer
reuse each other's entry. Changing this setting means existing result level
cache entries are no longer read.|false|
Review Comment:
I generally dislike adding so many random config flags, but given this could
substantially negatively impact cache hit rates so seems to me.
##########
server/src/test/java/org/apache/druid/query/ResultLevelCachingQueryRunnerTest.java:
##########
@@ -404,6 +405,97 @@ public void testPopulateCacheThrowsException()
emitter.verifyValue(DefaultQueryMetrics.QUERY_RESULT_CACHE_HIT, 0);
}
+ @Test
Review Comment:
non-blocking nit: might be nice to add a test that verifies the case where a
cache is persistent across a process restart (i.e external cache)? And that
populating with the flag off -> enable it -> change segments -> disable it
again w/ assert the old entry is rejected and current results are returned.
##########
server/src/main/java/org/apache/druid/query/ResultLevelCachingQueryRunner.java:
##########
@@ -206,6 +209,29 @@ private byte[] fetchResultsFromResultLevelCache(
return null;
}
+ private byte[] computeResultLevelCacheKey()
+ {
+ final byte[] queryCacheKey = strategy.computeResultLevelCacheKey(query);
+ if (!cacheConfig.isIncludeDatasourceAndIntervalsInResultLevelCacheKey()) {
+ return queryCacheKey;
+ }
+ // The cached value still starts with the ETag, which is checked before
the entry is used. Adding the datasource
+ // and intervals to the key gives queries that differ only by interval or
datasource their own entries, instead of
+ // one shared entry that each of them overwrites. The byte arrays are
length-prefixed so that where one ends and the
+ // next begins is unambiguous.
+ final CacheKeyBuilder builder = new
CacheKeyBuilder(RESULT_LEVEL_CACHE_KEY_WITH_DATASOURCE_AND_INTERVALS)
+ .appendInt(queryCacheKey.length)
+ .appendByteArray(queryCacheKey);
+ final byte[] dataSourceCacheKey = query.getDataSource().getCacheKey();
+ if (dataSourceCacheKey != null) {
+
builder.appendInt(dataSourceCacheKey.length).appendByteArray(dataSourceCacheKey);
+ }
+ for (Interval interval : query.getIntervals()) {
Review Comment:
getIntervals() is sorted but might be nice to add a small test to ensure we
get equal cache keys queries which pass different order intervals :)
and might be worth thinking about appendLongArray() or similar if it exists,
looks like that might shave off some bytes on the cache key size
##########
server/src/test/java/org/apache/druid/query/ResultLevelCachingQueryRunnerTest.java:
##########
@@ -404,6 +405,97 @@ public void testPopulateCacheThrowsException()
emitter.verifyValue(DefaultQueryMetrics.QUERY_RESULT_CACHE_HIT, 0);
}
+ @Test
+ public void testDifferentIntervalsShareOneEntryByDefault()
+ {
+ prepareCluster(10);
+ final Query<Result<TimeseriesResultValue>> query1 =
timeseriesQuery(Intervals.of("2000-01-01T00/PT2H"));
+ final Query<Result<TimeseriesResultValue>> query2 =
timeseriesQuery(Intervals.of("2000-01-01T01/PT2H"));
+ final CacheConfig config = newCacheConfig(true, true,
DEFAULT_CACHE_ENTRY_MAX_SIZE);
+
+ createQueryRunner(config, query1).run(QueryPlus.wrap(query1),
responseContext()).toList();
+ createQueryRunner(config, query2).run(QueryPlus.wrap(query2),
responseContext()).toList();
+ Assertions.assertEquals(1, cache.getStats().getNumEntries());
+ emitter.flush();
+
+ createQueryRunner(config, query1).run(QueryPlus.wrap(query1),
responseContext()).toList();
+
+ // query2 overwrote query1's entry, so the last run finds an entry (a hit
for the cache) but can't use it.
+ Assertions.assertEquals(1, cache.getStats().getNumEntries());
+ Assertions.assertEquals(2, cache.getStats().getNumHits());
+ emitter.verifyEmitted(DefaultQueryMetrics.QUERY_RESULT_CACHE_HIT, 1);
+ emitter.verifyValue(DefaultQueryMetrics.QUERY_RESULT_CACHE_HIT, 0);
+ }
+
+ @Test
+ public void testDifferentIntervalsUseSeparateEntriesWhenFlagEnabled()
+ {
+ prepareCluster(10);
+ final Query<Result<TimeseriesResultValue>> query1 =
timeseriesQuery(Intervals.of("2000-01-01T00/PT2H"));
+ final Query<Result<TimeseriesResultValue>> query2 =
timeseriesQuery(Intervals.of("2000-01-01T01/PT2H"));
+ final CacheConfig config = newCacheConfig(true, true,
DEFAULT_CACHE_ENTRY_MAX_SIZE, true);
+
+ final List<Result<TimeseriesResultValue>> results1 =
+ createQueryRunner(config, query1).run(QueryPlus.wrap(query1),
responseContext()).toList();
+ final List<Result<TimeseriesResultValue>> results2 =
+ createQueryRunner(config, query2).run(QueryPlus.wrap(query2),
responseContext()).toList();
+ Assertions.assertEquals(2, cache.getStats().getNumEntries());
+ emitter.flush();
+
+ Assertions.assertEquals(
+ results1,
+ createQueryRunner(config, query1).run(QueryPlus.wrap(query1),
responseContext()).toList()
+ );
+ emitter.verifyEmitted(DefaultQueryMetrics.QUERY_RESULT_CACHE_HIT, 1);
+ emitter.verifyValue(DefaultQueryMetrics.QUERY_RESULT_CACHE_HIT, 1);
+ emitter.flush();
+
+ Assertions.assertEquals(
+ results2,
+ createQueryRunner(config, query2).run(QueryPlus.wrap(query2),
responseContext()).toList()
+ );
+ emitter.verifyEmitted(DefaultQueryMetrics.QUERY_RESULT_CACHE_HIT, 1);
+ emitter.verifyValue(DefaultQueryMetrics.QUERY_RESULT_CACHE_HIT, 1);
+ Assertions.assertEquals(2, cache.getStats().getNumEntries());
+ }
+
+ @Test
+ public void testDatasourceAndIntervalsInCacheKeyOnlyWhenFlagEnabled()
+ {
+ cache = Mockito.spy(cache);
+ prepareCluster(10);
+ final Query<Result<TimeseriesResultValue>> query =
timeseriesQuery(Intervals.of("2000-01-01T00/PT2H"));
+ final Query<Result<TimeseriesResultValue>> sameQuery =
timeseriesQuery(Intervals.of("2000-01-01T00/PT2H"));
+ final Query<Result<TimeseriesResultValue>> otherInterval =
timeseriesQuery(Intervals.of("2000-01-01T01/PT2H"));
Review Comment:
maybe add another overlap case: `"2000-01-01T00/PT2H","2000-01-01T02/PT2H"`
and `"2000-01-01T00/PT2H","2000-01-01T03/PT2H"`
##########
server/src/test/java/org/apache/druid/query/ResultLevelCachingQueryRunnerTest.java:
##########
@@ -404,6 +405,97 @@ public void testPopulateCacheThrowsException()
emitter.verifyValue(DefaultQueryMetrics.QUERY_RESULT_CACHE_HIT, 0);
}
+ @Test
+ public void testDifferentIntervalsShareOneEntryByDefault()
+ {
+ prepareCluster(10);
+ final Query<Result<TimeseriesResultValue>> query1 =
timeseriesQuery(Intervals.of("2000-01-01T00/PT2H"));
Review Comment:
nit: let's test same start but different end and same end but different
start?
##########
server/src/main/java/org/apache/druid/query/ResultLevelCachingQueryRunner.java:
##########
@@ -95,7 +98,7 @@ public Sequence<T> run(QueryPlus queryPlus, ResponseContext
responseContext)
{
if (useResultCache || populateResultCache) {
- final byte[] queryCacheKey = strategy.computeResultLevelCacheKey(query);
+ final byte[] queryCacheKey = computeResultLevelCacheKey();
final Cache.NamedKey cacheKey =
CacheUtil.computeResultLevelCacheKey(queryCacheKey);
final byte[] cachedResultSet =
fetchResultsFromResultLevelCache(cacheKey);
Review Comment:
something I noticed while looking over
[this](https://github.com/maytasm/druid/blob/f577780e1c7d044cddda8e629c3e8a1dadbae678/server/src/main/java/org/apache/druid/query/ResultLevelCachingQueryRunner.java#L366)
code is that looks like there might? be a further way to increase hit rate.
Consider 3 query instances that map to the same query cache key:
```
q1: read cache, calculate E1 from old segments, start execution
q2: read cache, calculate E2 from new segments, start execution
q2: finish, write (key, {E2, results2})
q1: finish, write (key, {E1, results1})
q3: read cached {E1, results1}
q3: calculate E2 from current segments
q3: reject E1, execute query, write (key, {E2, results3})
```
In order to prevent q3 from executing, we might be able to either:
- Do some sort of cache.putIfAbsentOrOlder() which sets the result if it is
not there yet or the etag represents a segment set that is "fresher" than the
prior cached result.
- Alternatively, can maybe think about storing the etag in the cache key
itself which would blow up the cache size potentially (and you'd need eviction
policy which is complicated)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
