Query DSL: term, term, prefix, and range filter are now weakly cached, for more strong caching, set _cache to true, closes elastic#450.

Author: kimchy

modules/elasticsearch/src/main/java/org/apache/lucene/search/PublicTermsFilter.java

@@ -19,8 +19,12 @@
 
 package org.apache.lucene.search;
 
+import org.apache.lucene.index.IndexReader;
 import org.apache.lucene.index.Term;
+import org.apache.lucene.index.TermDocs;
+import org.apache.lucene.util.OpenBitSet;
 
+import java.io.IOException;
 import java.util.Set;
 
 /**
@@ -31,4 +35,24 @@ public class PublicTermsFilter extends TermsFilter {
     public Set<Term> getTerms() {
         return terms;
     }
+
+    // override default impl here to use fastSet...
+
+    @Override
+    public DocIdSet getDocIdSet(IndexReader reader) throws IOException {
+        OpenBitSet result = new OpenBitSet(reader.maxDoc());
+        TermDocs td = reader.termDocs();
+        try {
+            for (Term term : terms) {
+                td.seek(term);
+                while (td.next()) {
+                    result.fastSet(td.doc());
+                }
+            }
+        }
+        finally {
+            td.close();
+        }
+        return result;
+    }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/common/lucene/search/TermFilter.java

@@ -31,7 +31,7 @@
 /**
  * A simple filter for a specific term.
  *
- * @author kimchy (Shay Banon)
+ * @author kimchy (shay.banon)
  */
 public class TermFilter extends Filter {
 
@@ -51,7 +51,7 @@ public Term getTerm() {
         try {
             td.seek(term);
             while (td.next()) {
-                result.set(td.doc());
+                result.fastSet(td.doc());
             }
         }
         finally {

modules/elasticsearch/src/main/java/org/elasticsearch/index/cache/filter/FilterCache.java

@@ -33,6 +33,8 @@ public interface FilterCache extends IndexComponent, CloseableComponent {
 
     Filter cache(Filter filterToCache);
 
+    Filter weakCache(Filter filterToCache);
+
     boolean isCached(Filter filter);
 
     void clear(IndexReader reader);

modules/elasticsearch/src/main/java/org/elasticsearch/index/cache/filter/none/NoneFilterCache.java

@@ -50,6 +50,10 @@ public class NoneFilterCache extends AbstractIndexComponent implements FilterCac
         return filterToCache;
     }
 
+    @Override public Filter weakCache(Filter filterToCache) {
+        return filterToCache;
+    }
+
     @Override public boolean isCached(Filter filter) {
         return false;
     }

modules/elasticsearch/src/main/java/org/elasticsearch/index/cache/filter/soft/SoftFilterCache.java

@@ -25,7 +25,7 @@
 import org.elasticsearch.common.lucene.docset.DocSet;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.Index;
-import org.elasticsearch.index.cache.filter.support.AbstractConcurrentMapFilterCache;
+import org.elasticsearch.index.cache.filter.support.AbstractDoubleConcurrentMapFilterCache;
 import org.elasticsearch.index.settings.IndexSettings;
 
 import java.util.concurrent.ConcurrentMap;
@@ -35,18 +35,24 @@
  *
  * @author kimchy (shay.banon)
  */
-public class SoftFilterCache extends AbstractConcurrentMapFilterCache {
+public class SoftFilterCache extends AbstractDoubleConcurrentMapFilterCache {
 
     @Inject public SoftFilterCache(Index index, @IndexSettings Settings indexSettings) {
         super(index, indexSettings);
     }
 
-    @Override protected ConcurrentMap<Filter, DocSet> buildFilterMap() {
+    @Override protected ConcurrentMap<Filter, DocSet> buildCacheMap() {
         // DocSet are not really stored with strong reference only when searching on them...
         // Filter might be stored in query cache
         return new MapMaker().softValues().makeMap();
     }
 
+    @Override protected ConcurrentMap<Filter, DocSet> buildWeakCacheMap() {
+        // DocSet are not really stored with strong reference only when searching on them...
+        // Filter might be stored in query cache
+        return new MapMaker().weakValues().makeMap();
+    }
+
     @Override public String type() {
         return "soft";
     }

modules/elasticsearch/src/main/java/org/elasticsearch/index/cache/filter/support/AbstractConcurrentMapFilterCache.java

@@ -95,6 +95,10 @@ protected AbstractConcurrentMapFilterCache(Index index, @IndexSettings Settings
         return new FilterCacheFilterWrapper(filterToCache, this);
     }
 
+    @Override public Filter weakCache(Filter filterToCache) {
+        return cache(filterToCache);
+    }
+
     @Override public boolean isCached(Filter filter) {
         return filter instanceof FilterCacheFilterWrapper;
     }

modules/elasticsearch/src/main/java/org/elasticsearch/index/cache/filter/support/AbstractDoubleConcurrentMapFilterCache.java

@@ -0,0 +1,221 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.index.cache.filter.support;
+
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.search.DocIdSet;
+import org.apache.lucene.search.Filter;
+import org.elasticsearch.common.collect.MapMaker;
+import org.elasticsearch.common.lucene.docset.DocSet;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.AbstractIndexComponent;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.index.cache.filter.FilterCache;
+import org.elasticsearch.index.settings.IndexSettings;
+
+import java.io.IOException;
+import java.util.concurrent.ConcurrentMap;
+
+import static org.elasticsearch.common.lucene.docset.DocSets.*;
+import static org.elasticsearch.common.util.concurrent.ConcurrentCollections.*;
+
+/**
+ * A base concurrent filter cache that accepts the actual cache to use.
+ *
+ * @author kimchy (shay.banon)
+ */
+public abstract class AbstractDoubleConcurrentMapFilterCache extends AbstractIndexComponent implements FilterCache {
+
+    final ConcurrentMap<Object, ConcurrentMap<Filter, DocSet>> cache;
+    final ConcurrentMap<Object, ConcurrentMap<Filter, DocSet>> weakCache;
+
+    protected AbstractDoubleConcurrentMapFilterCache(Index index, @IndexSettings Settings indexSettings) {
+        super(index, indexSettings);
+        // weak keys is fine, it will only be cleared once IndexReader references will be removed
+        // (assuming clear(...) will not be called)
+        this.cache = new MapMaker().weakKeys().makeMap();
+        this.weakCache = new MapMaker().weakKeys().makeMap();
+    }
+
+    @Override public void close() {
+        cache.clear();
+    }
+
+    @Override public void clear() {
+        cache.clear();
+    }
+
+    @Override public void clear(IndexReader reader) {
+        ConcurrentMap<Filter, DocSet> map = cache.remove(reader.getFieldCacheKey());
+        // help soft/weak handling GC
+        if (map != null) {
+            map.clear();
+        }
+        map = weakCache.remove(reader.getFieldCacheKey());
+        // help soft/weak handling GC
+        if (map != null) {
+            map.clear();
+        }
+    }
+
+    @Override public void clearUnreferenced() {
+    }
+
+    @Override public Filter cache(Filter filterToCache) {
+        if (isCached(filterToCache)) {
+            return filterToCache;
+        }
+        return new FilterCacheFilterWrapper(filterToCache, this);
+    }
+
+    @Override public Filter weakCache(Filter filterToCache) {
+        if (isCached(filterToCache)) {
+            return filterToCache;
+        }
+        return new FilterWeakCacheFilterWrapper(filterToCache, this);
+    }
+
+    @Override public boolean isCached(Filter filter) {
+        return filter instanceof CacheMarker;
+    }
+
+    protected ConcurrentMap<Filter, DocSet> buildCacheMap() {
+        return newConcurrentMap();
+    }
+
+    protected ConcurrentMap<Filter, DocSet> buildWeakCacheMap() {
+        return newConcurrentMap();
+    }
+
+    static abstract class CacheMarker extends Filter {
+
+    }
+
+    // LUCENE MONITOR: Check next version Lucene for CachingWrapperFilter, consider using that logic
+    // and not use the DeletableConstantScoreQuery, instead pass the DeletesMode enum to the cache method
+    // see: https://issues.apache.org/jira/browse/LUCENE-2468
+
+    static class FilterCacheFilterWrapper extends CacheMarker {
+
+        private final Filter filter;
+
+        private final AbstractDoubleConcurrentMapFilterCache cache;
+
+        FilterCacheFilterWrapper(Filter filter, AbstractDoubleConcurrentMapFilterCache cache) {
+            this.filter = filter;
+            this.cache = cache;
+        }
+
+        @Override public DocIdSet getDocIdSet(IndexReader reader) throws IOException {
+            ConcurrentMap<Filter, DocSet> cachedFilters = cache.cache.get(reader.getFieldCacheKey());
+            if (cachedFilters == null) {
+                cachedFilters = cache.buildCacheMap();
+                cache.cache.putIfAbsent(reader.getFieldCacheKey(), cachedFilters);
+            }
+            DocSet docSet = cachedFilters.get(filter);
+            if (docSet != null) {
+                return docSet;
+            }
+
+            // check if its in the weak cache, if so, move it from weak to soft
+            ConcurrentMap<Filter, DocSet> weakCachedFilters = cache.weakCache.get(reader.getFieldCacheKey());
+            if (weakCachedFilters != null) {
+                docSet = weakCachedFilters.get(filter);
+                if (docSet != null) {
+                    cachedFilters.put(filter, docSet);
+                    weakCachedFilters.remove(filter);
+                    return docSet;
+                }
+            }
+
+            DocIdSet docIdSet = filter.getDocIdSet(reader);
+            docSet = cacheable(reader, docIdSet);
+            cachedFilters.putIfAbsent(filter, docSet);
+            return docIdSet;
+        }
+
+        public String toString() {
+            return "FilterCacheFilterWrapper(" + filter + ")";
+        }
+
+        public boolean equals(Object o) {
+            if (!(o instanceof FilterCacheFilterWrapper)) return false;
+            return this.filter.equals(((FilterCacheFilterWrapper) o).filter);
+        }
+
+        public int hashCode() {
+            return filter.hashCode() ^ 0x1117BF25;
+        }
+    }
+
+    static class FilterWeakCacheFilterWrapper extends CacheMarker {
+
+        private final Filter filter;
+
+        private final AbstractDoubleConcurrentMapFilterCache cache;
+
+        FilterWeakCacheFilterWrapper(Filter filter, AbstractDoubleConcurrentMapFilterCache cache) {
+            this.filter = filter;
+            this.cache = cache;
+        }
+
+        @Override public DocIdSet getDocIdSet(IndexReader reader) throws IOException {
+            DocSet docSet;
+            // first check if its in the actual cache
+            ConcurrentMap<Filter, DocSet> cachedFilters = cache.cache.get(reader.getFieldCacheKey());
+            if (cachedFilters != null) {
+                docSet = cachedFilters.get(filter);
+                if (docSet != null) {
+                    return docSet;
+                }
+            }
+
+            // now, handle it in the weak cache
+            ConcurrentMap<Filter, DocSet> weakCacheFilters = cache.weakCache.get(reader.getFieldCacheKey());
+            if (weakCacheFilters == null) {
+                weakCacheFilters = cache.buildWeakCacheMap();
+                cache.weakCache.putIfAbsent(reader.getFieldCacheKey(), weakCacheFilters);
+            }
+
+            docSet = weakCacheFilters.get(filter);
+            if (docSet != null) {
+                return docSet;
+            }
+
+            DocIdSet docIdSet = filter.getDocIdSet(reader);
+            docSet = cacheable(reader, docIdSet);
+            weakCacheFilters.putIfAbsent(filter, docSet);
+            return docIdSet;
+        }
+
+        public String toString() {
+            return "FilterCacheFilterWrapper(" + filter + ")";
+        }
+
+        public boolean equals(Object o) {
+            if (!(o instanceof FilterCacheFilterWrapper)) return false;
+            return this.filter.equals(((FilterCacheFilterWrapper) o).filter);
+        }
+
+        public int hashCode() {
+            return filter.hashCode() ^ 0x1117BF25;
+        }
+    }
+}

modules/elasticsearch/src/main/java/org/elasticsearch/index/query/xcontent/PrefixFilterParser.java

@@ -53,7 +53,7 @@ public class PrefixFilterParser extends AbstractIndexComponent implements XConte
     @Override public Filter parse(QueryParseContext parseContext) throws IOException, QueryParsingException {
         XContentParser parser = parseContext.parser();
 
-        boolean cache = true; // default to true, make sense to cache prefix
+        boolean cache = false;
         String fieldName = null;
         String value = null;
 
@@ -88,9 +88,15 @@ public class PrefixFilterParser extends AbstractIndexComponent implements XConte
         }
 
         Filter filter = new PrefixFilter(new Term(fieldName, value));
+
+        // we weak cache the filter if not cached, since in any case it builds an OpenBitSet
+        // we might as well weak cache it...
         if (cache) {
             filter = parseContext.cacheFilter(filter);
+        } else {
+            filter = parseContext.cacheWeakFilter(filter);
         }
+
         filter = wrapSmartNameFilter(filter, smartNameFieldMappers, parseContext);
         if (filterName != null) {
             parseContext.addNamedFilter(filterName, filter);

modules/elasticsearch/src/main/java/org/elasticsearch/index/query/xcontent/QueryParseContext.java

@@ -112,6 +112,10 @@ public Filter cacheFilter(Filter filter) {
         return indexQueryParser.indexCache.filter().cache(filter);
     }
 
+    public Filter cacheWeakFilter(Filter filter) {
+        return indexQueryParser.indexCache.filter().weakCache(filter);
+    }
+
     public void addNamedFilter(String name, Filter filter) {
         namedFilters.put(name, filter);
     }