samarthjain commented on code in PR #18568:
URL: https://github.com/apache/druid/pull/18568#discussion_r2392381071
##########
processing/src/main/java/org/apache/druid/query/scan/ScanQueryQueryToolChest.java:
##########
@@ -191,6 +198,80 @@ public Sequence<Object[]> resultsAsArrays(final ScanQuery
query, final Sequence<
);
}
+ @Override
+ public CacheStrategy<ScanResultValue, ScanResultValue, ScanQuery>
getCacheStrategy(
+ final ScanQuery query,
+ @Nullable final ObjectMapper objectMapper
+ )
+ {
+ return new CacheStrategy<>()
+ {
+ @Override
+ public boolean isCacheable(ScanQuery query, boolean willMergeRunners,
boolean segmentLevel)
+ {
+ // Currently, there is no bijective mapping from ScanResultValue to
Result<BySegmentResultValueClass<ScanResultValue>>.
+ // This means queries will fail if:
+ // - A query is issued with bySegment:true
+ // - Segment-level cache is enabled on the broker (in which case it
sends bySegment queries to data nodes).
+ return !query.context().isBySegment() && (!segmentLevel ||
willMergeRunners);
+ }
+
+ @Override
+ public byte[] computeCacheKey(ScanQuery query)
+ {
+ CacheKeyBuilder builder = new CacheKeyBuilder(SCAN_QUERY)
+ .appendCacheable(query.getVirtualColumns())
+ .appendString(query.getResultFormat().toString())
Review Comment:
Do you know if resultFormat is used after getting results out of the result
set cache or before the results are return to the cache? If former, we probably
can remove it from the cache key.
##########
processing/src/main/java/org/apache/druid/query/scan/ScanQueryQueryToolChest.java:
##########
@@ -191,6 +198,80 @@ public Sequence<Object[]> resultsAsArrays(final ScanQuery
query, final Sequence<
);
}
+ @Override
+ public CacheStrategy<ScanResultValue, ScanResultValue, ScanQuery>
getCacheStrategy(
+ final ScanQuery query,
+ @Nullable final ObjectMapper objectMapper
+ )
+ {
+ return new CacheStrategy<>()
+ {
+ @Override
+ public boolean isCacheable(ScanQuery query, boolean willMergeRunners,
boolean segmentLevel)
+ {
+ // Currently, there is no bijective mapping from ScanResultValue to
Result<BySegmentResultValueClass<ScanResultValue>>.
+ // This means queries will fail if:
+ // - A query is issued with bySegment:true
+ // - Segment-level cache is enabled on the broker (in which case it
sends bySegment queries to data nodes).
+ return !query.context().isBySegment() && (!segmentLevel ||
willMergeRunners);
+ }
+
+ @Override
+ public byte[] computeCacheKey(ScanQuery query)
+ {
+ CacheKeyBuilder builder = new CacheKeyBuilder(SCAN_QUERY)
+ .appendCacheable(query.getVirtualColumns())
Review Comment:
I see `GroupByQueryToolChest` has a CACHE_STRATEGY_VERSION. I suppose the
intent of having versioning is to discard old data from the cache if the cache
key structure changes. It would make sense to add it here as well to ensure key
evolution doesn't cause any unpleasant gotchas in the future.
##########
processing/src/main/java/org/apache/druid/query/scan/ScanQueryQueryToolChest.java:
##########
@@ -191,6 +198,80 @@ public Sequence<Object[]> resultsAsArrays(final ScanQuery
query, final Sequence<
);
}
+ @Override
+ public CacheStrategy<ScanResultValue, ScanResultValue, ScanQuery>
getCacheStrategy(
+ final ScanQuery query,
+ @Nullable final ObjectMapper objectMapper
+ )
+ {
+ return new CacheStrategy<>()
+ {
+ @Override
+ public boolean isCacheable(ScanQuery query, boolean willMergeRunners,
boolean segmentLevel)
+ {
+ // Currently, there is no bijective mapping from ScanResultValue to
Result<BySegmentResultValueClass<ScanResultValue>>.
+ // This means queries will fail if:
+ // - A query is issued with bySegment:true
+ // - Segment-level cache is enabled on the broker (in which case it
sends bySegment queries to data nodes).
+ return !query.context().isBySegment() && (!segmentLevel ||
willMergeRunners);
+ }
+
+ @Override
+ public byte[] computeCacheKey(ScanQuery query)
+ {
+ CacheKeyBuilder builder = new CacheKeyBuilder(SCAN_QUERY)
+ .appendCacheable(query.getVirtualColumns())
+ .appendString(query.getResultFormat().toString())
+ .appendInt(query.getBatchSize())
+ .appendLong(query.getScanRowsOffset())
+ .appendLong(query.getScanRowsLimit())
+ .appendCacheable(query.getFilter())
+ .appendStrings(query.getColumns() != null ? query.getColumns() :
List.of())
+ .appendString(query.getTimeOrder().toString());
+
+ if (query.getOrderBys() != null && !query.getOrderBys().isEmpty()) {
Review Comment:
nit: the empty check can be removed since the iterator in the loop will
check that internally.
```
List<OrderBy> orderBys = query.getOrderBys();
if (orderBys != null) {
for (OrderBy orderBy : orderBys) {
builder.appendString(orderBy.getColumnName())
.appendString(orderBy.getOrder().toString());
}
}
```
##########
processing/src/test/java/org/apache/druid/query/scan/ScanQueryQueryToolChestTest.java:
##########
@@ -441,4 +447,247 @@ private static Sequence<ScanResultValue> results3()
)
);
}
+
+ @Test
+ public void testCacheStrategy()
+ {
+ ScanQuery query = Druids.newScanQueryBuilder()
+ .dataSource("foo")
+ .intervals(new
MultipleIntervalSegmentSpec(ImmutableList.of(Intervals.of("2015-01-01/2015-01-02"))))
+ .columns("dim1", "dim2")
+ .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST)
+ .batchSize(4096)
+ .offset(10)
+ .limit(100)
+ .build();
+
+ CacheStrategy<ScanResultValue, ScanResultValue, ScanQuery> strategy =
toolChest.getCacheStrategy(query, null);
+
+ Assert.assertNotNull(strategy);
+ Assert.assertTrue(strategy.isCacheable(query, true, false));
+ Assert.assertFalse(strategy.isCacheable(query, false, true));
+ Assert.assertTrue(strategy.isCacheable(query, true, true));
+
+ byte[] cacheKey = strategy.computeCacheKey(query);
+ Assert.assertNotNull(cacheKey);
+ Assert.assertTrue(cacheKey.length > 0);
+
+ byte[] resultLevelCacheKey = strategy.computeResultLevelCacheKey(query);
+ Assert.assertNotNull(resultLevelCacheKey);
+ Assert.assertTrue(resultLevelCacheKey.length > 0);
+
+ // For ScanQuery, result-level and segment-level cache keys should be the
same
+ Assert.assertArrayEquals(cacheKey, resultLevelCacheKey);
+
+ ScanResultValue testResult = new ScanResultValue(
+ "test_segment",
+ ImmutableList.of("dim1", "dim2"),
+ ImmutableList.of(
+ ImmutableMap.of("dim1", "value1", "dim2", "value2"),
+ ImmutableMap.of("dim1", "value3", "dim2", "value4")
+ )
+ );
+
+ ScanResultValue cachedValue =
strategy.prepareForCache(false).apply(testResult);
+ ScanResultValue fromCache =
strategy.pullFromCache(false).apply(cachedValue);
+
+ Assert.assertEquals(testResult, fromCache);
+ }
+
+ @Test
+ public void testCacheDisabledForBySegmentQueries()
+ {
+ ScanQuery query = Druids.newScanQueryBuilder()
+ .dataSource("foo")
+ .intervals(new
MultipleIntervalSegmentSpec(ImmutableList.of(Intervals.of("2015-01-01/2015-01-02"))))
+ .columns("dim1", "dim2")
+
.resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST)
+ .batchSize(4096)
+ .offset(10)
+ .limit(100)
+ .context(ImmutableMap.of("bySegment", true))
+ .build();
+
+ CacheStrategy<ScanResultValue, ScanResultValue, ScanQuery> strategy =
toolChest.getCacheStrategy(query, null);
+
+ Assert.assertNotNull(strategy);
+ Assert.assertFalse(strategy.isCacheable(query, true, false));
+ Assert.assertFalse(strategy.isCacheable(query, false, true));
+ }
+
+ @Test
+ public void testCacheKeyDifferentQueries()
+ {
+ ScanQuery query1 = Druids.newScanQueryBuilder()
+ .dataSource("foo")
+ .intervals(new
MultipleIntervalSegmentSpec(ImmutableList.of(Intervals.of("2025-01-01/2025-01-02"))))
+ .columns("dim1", "dim2")
+ .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST)
+ .build();
+
+ ScanQuery query2 = Druids.newScanQueryBuilder()
+ .dataSource("foo")
+ .intervals(new
MultipleIntervalSegmentSpec(ImmutableList.of(Intervals.of("2025-01-01/2025-01-02"))))
+ .columns("dim1", "dim3")
+ .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST)
+ .build();
+
+ ScanQuery query3 = Druids.newScanQueryBuilder()
Review Comment:
Just like you have tests for other fields, it might make sense to separate
out the test for result format on its own. Also, we may or may need to have
result format as the cache key in which case this test may not be needed.
##########
processing/src/main/java/org/apache/druid/query/scan/ScanQueryQueryToolChest.java:
##########
@@ -191,6 +198,80 @@ public Sequence<Object[]> resultsAsArrays(final ScanQuery
query, final Sequence<
);
}
+ @Override
+ public CacheStrategy<ScanResultValue, ScanResultValue, ScanQuery>
getCacheStrategy(
+ final ScanQuery query,
+ @Nullable final ObjectMapper objectMapper
+ )
+ {
+ return new CacheStrategy<>()
+ {
+ @Override
+ public boolean isCacheable(ScanQuery query, boolean willMergeRunners,
boolean segmentLevel)
+ {
+ // Currently, there is no bijective mapping from ScanResultValue to
Result<BySegmentResultValueClass<ScanResultValue>>.
+ // This means queries will fail if:
+ // - A query is issued with bySegment:true
+ // - Segment-level cache is enabled on the broker (in which case it
sends bySegment queries to data nodes).
+ return !query.context().isBySegment() && (!segmentLevel ||
willMergeRunners);
+ }
+
+ @Override
+ public byte[] computeCacheKey(ScanQuery query)
+ {
+ CacheKeyBuilder builder = new CacheKeyBuilder(SCAN_QUERY)
+ .appendCacheable(query.getVirtualColumns())
Review Comment:
Interestingly, I don't see datasource being used in the cache key. Is that
check performed somewhere else? It would be good to add a test like
`testCacheKeyDifferentQueries` where you have everything else the same but the
datasource names are different. If the test fails, I suppose we may have
discovered a latent bug :)
##########
processing/src/main/java/org/apache/druid/query/scan/ScanQueryQueryToolChest.java:
##########
@@ -191,6 +198,80 @@ public Sequence<Object[]> resultsAsArrays(final ScanQuery
query, final Sequence<
);
}
+ @Override
+ public CacheStrategy<ScanResultValue, ScanResultValue, ScanQuery>
getCacheStrategy(
+ final ScanQuery query,
+ @Nullable final ObjectMapper objectMapper
+ )
+ {
+ return new CacheStrategy<>()
+ {
+ @Override
+ public boolean isCacheable(ScanQuery query, boolean willMergeRunners,
boolean segmentLevel)
+ {
+ // Currently, there is no bijective mapping from ScanResultValue to
Result<BySegmentResultValueClass<ScanResultValue>>.
+ // This means queries will fail if:
+ // - A query is issued with bySegment:true
+ // - Segment-level cache is enabled on the broker (in which case it
sends bySegment queries to data nodes).
+ return !query.context().isBySegment() && (!segmentLevel ||
willMergeRunners);
+ }
+
+ @Override
+ public byte[] computeCacheKey(ScanQuery query)
+ {
+ CacheKeyBuilder builder = new CacheKeyBuilder(SCAN_QUERY)
+ .appendCacheable(query.getVirtualColumns())
+ .appendString(query.getResultFormat().toString())
+ .appendInt(query.getBatchSize())
+ .appendLong(query.getScanRowsOffset())
+ .appendLong(query.getScanRowsLimit())
+ .appendCacheable(query.getFilter())
+ .appendStrings(query.getColumns() != null ? query.getColumns() :
List.of())
+ .appendString(query.getTimeOrder().toString());
+
+ if (query.getOrderBys() != null && !query.getOrderBys().isEmpty()) {
+ for (OrderBy orderBy : query.getOrderBys()) {
+ builder.appendString(orderBy.getColumnName())
+ .appendString(orderBy.getOrder().toString());
+ }
+ }
+
+ if (query.getColumnTypes() != null &&
!query.getColumnTypes().isEmpty()) {
Review Comment:
Same as above.
--
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]
