populate cache without materializing results

This commit is contained in:
Xavier Léauté 2014-04-01 22:11:58 -07:00
parent 89dd0ee29c
commit ee34415470
1 changed files with 23 additions and 13 deletions

View File

@ -20,6 +20,8 @@
package io.druid.client; package io.druid.client;
import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Function;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists; import com.google.common.collect.Lists;
import com.metamx.common.guava.Accumulator; import com.metamx.common.guava.Accumulator;
import com.metamx.common.guava.Sequence; import com.metamx.common.guava.Sequence;
@ -33,6 +35,7 @@ import io.druid.query.QueryRunner;
import io.druid.query.QueryToolChest; import io.druid.query.QueryToolChest;
import io.druid.query.SegmentDescriptor; import io.druid.query.SegmentDescriptor;
import javax.annotation.Nullable;
import java.util.ArrayList; import java.util.ArrayList;
public class CachePopulatingQueryRunner<T> implements QueryRunner<T> public class CachePopulatingQueryRunner<T> implements QueryRunner<T>
@ -72,26 +75,33 @@ public class CachePopulatingQueryRunner<T> implements QueryRunner<T>
final boolean populateCache = Boolean.parseBoolean(query.getContextValue(CacheConfig.POPULATE_CACHE, "true")) final boolean populateCache = Boolean.parseBoolean(query.getContextValue(CacheConfig.POPULATE_CACHE, "true"))
&& strategy != null && strategy != null
&& cacheConfig.isPopulateCache() && cacheConfig.isPopulateCache();
// historical only populates distributed cache since the cache lookups are done at broker.
&& !(cache instanceof MapCache); final Sequence<T> results = base.run(query);
if (populateCache) { if (populateCache) {
Sequence<T> results = base.run(query); final Cache.NamedKey key = CacheUtil.computeSegmentCacheKey(
Cache.NamedKey key = CacheUtil.computeSegmentCacheKey(
segmentIdentifier, segmentIdentifier,
segmentDescriptor, segmentDescriptor,
strategy.computeCacheKey(query) strategy.computeCacheKey(query)
); );
ArrayList<T> resultAsList = Sequences.toList(results, new ArrayList<T>());
CacheUtil.populate( final Function cacheFn = strategy.prepareForCache();
cache, return Sequences.map(
mapper, results,
key, new Function<T, T>()
Lists.transform(resultAsList, strategy.prepareForCache()) {
@Nullable
@Override
public T apply(@Nullable T input)
{
CacheUtil.populate(cache, mapper, key, ImmutableList.of(cacheFn.apply(input)));
return input;
}
}
); );
return Sequences.simple(resultAsList);
} else { } else {
return base.run(query); return results;
} }
} }