Part2: Run Time-Consuming Solr Query Faster: Use Guava CacheBuilder to Cache Response

The Problem
In our web application, the very first request to solr server is a stats query. When there are more than 50 millions data, the first stats query may take 1, 2 or more minutes. As it need load millions of documents, terms into Solr. 

For subsequent stats queries, it will run faster as Solr load them into its caches, but it still takes 5 to 15 or more seconds as the stats query is a compute-intensive task, and there is too many data.

We need make it run faster to make the web GUI more responsive.
Main Steps
1. Auto run queries X minutes after no update after startup or commit to make the first stats query run faster
2.  Use Guava CacheBuilder to Cache Solr Response
This is described in this article.

Task: Use Guava CacheBuilder to Cache Solr Response
We would like to store response of time-consuming request into cache, sol later request will be much faster.

The Implementation
CacheManager
CacheManager is the key class in the implementation. The key of the outer ConcurrentHashMap is SolrCore, its value is a ConcurrentHashMap. The key of inner ConcurrentHashMap is cacheType: such as solr request. Its value is a Guava Cache.

By default the cache = CacheBuilder.newBuilder().concurrencyLevel(16).expireAfterAccess(10, TimeUnit.MINUTES).softValues().recordStats().build(); We can specify parameter -DcacheSpec=concurrencyLevel=10,expireAfterAccess=5m,softValues to use a different kind of cache.

It adds response to cache asynchronously.

public class CacheManager implements CacheStatsOpMXBean {
  protected static final Logger logger = LoggerFactory
      .getLogger(CacheManager.class);
  public static final String CACHE_TAG_SOLR_REQUEST = "CACHE_TAG_SOLR_REQUEST";
  @SuppressWarnings("rawtypes")
  private ConcurrentHashMap<SolrCore,ConcurrentHashMap<String,Cache>> cacheMap = new ConcurrentHashMap<SolrCore,ConcurrentHashMap<String,Cache>>();
  
  private static CacheManager instance = null;
  private ExecutorService executors;
  
  private static String cacheSpec;
  
  private CacheManager() {
    cacheSpec = System.getProperty("cacheSpec");
    executors = Executors.newCachedThreadPool();
  }
  
  public static CacheManager getInstance() {
    if (instance == null) {
      synchronized (CacheManager.class) {
        if (instance == null) {
          instance = new CacheManager();
        }
      }
    }
    return instance;
  }
  
  private <K,V> Cache<K,V> newCache() {
    Cache<K,V> result = null;
    if (StringUtils.isNotBlank(cacheSpec)) {
      try {
        result = CacheBuilder.from(cacheSpec).build();
      } catch (Exception e) {
        logger.error("Invalid cacheSpec: " + cacheSpec, e);
      }
    }
    if (result == null) {
      // default cache
      result = CacheBuilder.newBuilder().concurrencyLevel(16)
          .expireAfterAccess(10, TimeUnit.MINUTES).softValues()
          .recordStats().build();
    }
    return result;
  }
  
  public <K,V> Cache<K,V> getCache(SolrCore core, String cacheTag) {
    cacheMap.putIfAbsent(core, new ConcurrentHashMap<String,Cache>());
    ConcurrentHashMap<String,Cache> coreCache = cacheMap.get(core);
    coreCache.putIfAbsent(cacheTag, newCache());
    return coreCache.get(cacheTag);
  }
  
  public void invalidateAll(SolrCore core) {
    ConcurrentHashMap<String,Cache> coreCache = cacheMap.get(core);
    if (coreCache != null) {
      for (Cache cahe : coreCache.values()) {
        cahe.invalidateAll();
      }
    }
  }

  public void addToCache(final SolrCore core, final String cacheTag,
      final CacheKeySolrQueryRequest cacheKey, final Object rspObj) {
    executors.submit(new Runnable() {
      @Override
      public void run() {
        Cache<CacheKeySolrQueryRequest,Object> cache = CacheManager
            .getInstance().getCache(core, cacheTag);
        cache.put(cacheKey, rspObj);
      }
    });
  }
}

CacheKeySolrQueryRequest
We can't use SolrQueryRequest as the the key of Guava cache. Because it doesn't implement hashCode and equals methods.The hashCode would be different for different requests with same solr query, equals would be false.
So We extract params map: Map from SolrQueryRequest, and implements the hashCode and equals methods. The order in the map and String[] array doesn't matter.

We can also use the deepHahsCode and deepEquals from java-util.

public class CacheKeySolrQueryRequest implements Serializable {
  
  private static final long serialVersionUID = 1L;
  Map<String,String[]> paramsMap;
  String url;
  
  private CacheKeySolrQueryRequest(SolrQueryRequest request) {
    this.paramsMap = SolrParams.toMultiMap(request.getParams().toNamedList());
    // remove unimportant params
    paramsMap.remove(CommonParams.TIME_ALLOWED);
    if (request.getContext().get("url") != null) {
      this.url = request.getContext().get("url").toString();
    }
  }
  
  public static CacheKeySolrQueryRequest create(SolrQueryRequest request) {
    CacheKeySolrQueryRequest result = null;
    if ((request.getContentStreams() == null || !request.getContentStreams()
        .iterator().hasNext())) {
      result = new CacheKeySolrQueryRequest(request);
    }
    return result;    
  }

  public int hashCode() {
    final int prime = 31;
    int result = 1;
    result = prime * result + ((url == null) ? 0 : url.hashCode());
    // the order in the map doesn't matter
    if (paramsMap != null) {
      int mapHashCode = 1;
      for (Entry<String,String[]> entry : paramsMap.entrySet()) {
        mapHashCode = (entry.getKey() == null ? 0 : entry.getKey().hashCode());
        for (String value : entry.getValue()) {
          mapHashCode = prime * mapHashCode
              + (value == null ? 0 : value.hashCode());
        }
      }
      
      result = prime * result + mapHashCode;
    }
    return result;
  }

  public boolean equals(Object obj) {
    if (this == obj) return true;
    if (obj == null) return false;
    if (getClass() != obj.getClass()) return false;
    CacheKeySolrQueryRequest other = (CacheKeySolrQueryRequest) obj;
    if (url == null) {
      if (other.url != null) return false;
    } else if (!url.equals(other.url)) return false;
    
    if (paramsMap == null) {
      if (other.paramsMap != null) return false;
    } else {
      if (paramsMap.size() != other.paramsMap.size()) return false;
      
      Iterator<Entry<String,String[]>> it = paramsMap.entrySet().iterator();
      while (it.hasNext()) {
        Entry<String,String[]> entry = it.next();
        String[] thisValues = entry.getValue();
        String[] otherValues = other.paramsMap.get(entry.getKey());
        if (!haveSameElements(thisValues, otherValues)) return false;
      }
      if (it.hasNext()) {
        return false;
      }
    }
    return true;
  }
  
  // helper class, so we don't have to do a whole lot of autoboxing
  private static class Count {
    public int count = 0;
  }
  // from: http://stackoverflow.com/questions/13501142/java-arraylist-how-can-i-tell-if-two-lists-are-equal-order-not-mattering
  public boolean haveSameElements(String[] list1, String[] list2) {
    if (list1 == list2) return true;
    if (list1 == null || list2 == null || list1.length != list2.length) return false;
    HashMap<String,Count> counts = new HashMap<String,Count>();

    for (String item : list1) {
      if (!counts.containsKey(item)) counts.put(item, new Count());
      counts.get(item).count += 1;
    }
    for (String item : list2) {
      // If the map doesn't contain the item here, then this item wasn't in
      // list1
      if (!counts.containsKey(item)) return false;
      counts.get(item).count -= 1;
    }
    for (Map.Entry<String,Count> entry : counts.entrySet()) {
      if (entry.getValue().count != 0) return false;
    }
    return true;
  }  
}

ResponseCachedSearchHandler
If useCache is true, ResponseCachedSearchHandler will first try to load the response from the cache, if the response is already cached, it will return response directly. If this is the first time this request is executed, it will run the request, if the execution time is longer than minExecuteTime, put response into cache. by default minExecuteTime is -1, mean we will always put response into cache).
We can change value of minExecuteTime, so Solr will only cache response if the requests takes more than specified minimum time.

Before return cached response, we have to call oldRsp.setReturnFields(new SolrReturnFields(oldReq)); this will set what fields to return based on fl parameter in request. Otherwise, solr will return all fields: as no fl parameter is set.

Sub class can extend ResponseCachedSearchHandler: implement isUseCache() method to determine whether solr should cache the response; implement beforeReturnFromCache to do something before return cached response back to solr.

public class ResponseCachedSearchHandler extends SearchHandler {  
  protected static final String PARAM_USE_CACHE = "useCache",
      PARAM_MIN_EXECUTE_TIME = "minExecuteTime";
  
  protected boolean defUseCache = false;
  protected int defMinExecuteTime = -1;
  public void init(NamedList args) {
    super.init(args);
    if (args != null) {
      defUseCache = defaults.getBool(PARAM_USE_CACHE, false);
      defMinExecuteTime = defaults.getInt(PARAM_MIN_EXECUTE_TIME, -1);
    }
  }
  
  public void handleRequestBody(SolrQueryRequest oldReq,
      SolrQueryResponse oldRsp) throws Exception {
    
    boolean useCache = isUseCache(oldReq);
    CacheKeySolrQueryRequest cacheKey = null;
    if (useCache) {
      Cache<CacheKeySolrQueryRequest,Object> cache = CacheManager
          .getInstance().getCache(oldReq.getCore(),
              CacheManager.CACHE_TAG_SOLR_REQUEST);
      
      cacheKey = CacheKeySolrQueryRequest.create(oldReq);
      if (cacheKey != null) {
        Object cachedRsp = cache.getIfPresent(cacheKey);
        if (cachedRsp != null) {
          NamedList<Object> valuesNL = oldRsp.getValues();
          valuesNL.add("response", cachedRsp);
          // SolrReturnFields defines which fields to return.
          oldRsp.setReturnFields(new SolrReturnFields(oldReq));
          beforeReturnFromCache(oldReq, oldRsp);
          return;
        }
      }
    }
    Stopwatch stopwatch = new Stopwatch().start();
    executeRequest(oldReq, oldRsp);
    long executeTime = stopwatch.elapsedTime(TimeUnit.MILLISECONDS);
    stopwatch.stop();
    beforeReturnNoCache(oldReq, oldRsp);
    addRspToCache(oldReq, oldRsp, useCache, cacheKey, executeTime);
  }
  
  protected void addRspToCache(SolrQueryRequest oldReq,
      SolrQueryResponse oldRsp, boolean useCache,
      CacheKeySolrQueryRequest cacheKey, long executeTime) {
    long minExecuteTime = oldReq.getParams().getInt(PARAM_MIN_EXECUTE_TIME,
        defMinExecuteTime);
    if (useCache && cacheKey != null && executeTime > minExecuteTime) {
      NamedList<Object> valuesNL = oldRsp.getValues();
      Object rspObj = (Object) valuesNL.get("response");
      CacheManager.getInstance().addToCache(oldReq.getCore(),
          CacheManager.CACHE_TAG_SOLR_REQUEST, cacheKey, rspObj);      
    }
  }
  
  /**
   * SubClass can extend this to check whether the request is stats query etc.
   */
  protected boolean isUseCache(SolrQueryRequest oldReq) {
    return oldReq.getParams().getBool(PARAM_USE_CACHE, defUseCache);
  }
  
  protected void beforeReturnNoCache(SolrQueryRequest oldReq,
      SolrQueryResponse oldRsp) {}

  protected void beforeReturnFromCache(SolrQueryRequest oldReq,
      SolrQueryResponse oldRsp) {}
      
  /**
   * by default, call searchHander.executeRequest
   */
  protected void executeRequest(SolrQueryRequest oldReq,
      SolrQueryResponse oldRsp) throws Exception {
    super.handleRequestBody(oldReq, oldRsp);
  }
}

CacheStatsFacetRequestHandler
CacheStatsFacetRequestHandler extends ResponseCachedSearchHandler, so solr will only store response of stats and facet requests. We will change the default requestHandler to use CacheStatsFacetRequestHandler.

<requestHandler name="/select" class="CacheStatsFacetRequestHandler" default="true">
    <!-- omitted -->
  </requestHandler>

public class CacheStatsFacetRequestHandler extends ResponseCachedSearchHandler {
  protected boolean isUseCache(SolrQueryRequest oldReq) {
    boolean useCache = super.isUseCache(oldReq);
    if (useCache) {
      SolrParams params = oldReq.getParams();
      useCache = params.getBool(StatsParams.STATS, false)
          || params.getBool(FacetParams.FACET, false);
    }
    return useCache;
  }
}
InvalidateCacheProcessorFactory
We need invalidate caches after solr commit. We need add the InvalidateCacheProcessorFactory to the default processor chain, and every updateRequestProcessorChain.
<updateRequestProcessorChain name="defaultChain" default="true">
    <processor class="solr.LogUpdateProcessorFactory" />
    <processor class="solr.RunUpdateProcessorFactory" />
    <processor class="InvalidateCacheProcessorFactory" />
    <processor
        class="AutoRunQueriesProcessorFactory"/>      
  </updateRequestProcessorChain>
public class InvalidateCacheProcessorFactory extends
    UpdateRequestProcessorFactory {
  public UpdateRequestProcessor getInstance(SolrQueryRequest req,
      SolrQueryResponse rsp, UpdateRequestProcessor next) {
    return new InvalidateCacheProcessor(next);
  }  
  private static class InvalidateCacheProcessor extends
      UpdateRequestProcessor {    
    public InvalidateCacheProcessor(UpdateRequestProcessor next) {
      super(next);
    }
    public void processCommit(CommitUpdateCommand cmd) throws IOException {
      super.processCommit(cmd);
      CacheManager.getInstance().invalidateAll(cmd.getReq().getCore());
    }
  }
}