深入源码分析 Elasticsearch Filter 机制

矫情吗；* 2022-11-22 12:59 174阅读 0赞

Elasticsearch Search Filter有一套和Query截然不同的运行机制，合理的运用Filter能够有效的提高检索的执行效率。本篇博客我将带领大家从源码角度深入理解Elasticsearch Search Filter的初始化，运行机制，注意事项，对比优势等等关键要点，希望能够让大家对Filter有更深的理解，能更好的运用。

### Filter Cache实例化 ###

Elasticsearch的Node在实例化时，会new一个`IndicesService`，其在构造函数中实例化了一个 `IndicesQueryCache`

#### ES IndicesQueryCache ElasticsearchLRUQueryCache 实例化 ####

public class IndicesQueryCache implements QueryCache, Closeable { 
    	// Cache内存占用空间上限
        public static final Setting<ByteSizeValue> INDICES_CACHE_QUERY_SIZE_SETTING = 
               Setting.memorySizeSetting("indices.queries.cache.size", "10%", Property.NodeScope);
        // Cache 元素个数上限
       	public static final Setting<Integer> INDICES_CACHE_QUERY_COUNT_SETTING = 
               Setting.intSetting("indices.queries.cache.count", 10_000, 1, Property.NodeScope);
    	// LRU形式的cache
    	private final LRUQueryCache cache;
    
    	public IndicesQueryCache(Settings settings) { 
            final ByteSizeValue size = INDICES_CACHE_QUERY_SIZE_SETTING.get(settings);
            final int count = INDICES_CACHE_QUERY_COUNT_SETTING.get(settings);
            logger.debug("using [node] query cache with size [{}] max filter count [{}]",
                    size, count);
            // 实例化LRU Cache, 上限10000个元素或者10%的堆内存大小
            cache = new ElasticsearchLRUQueryCache(count, size.getBytes());
            sharedRamBytesUsed = 0;
        }
    }

IndicesQueryCache 在实例化时会指定内存空间上限以及元素个数上限，通过这两个指标来达到LRU缓存淘汰的效果。这两个指标是`Node级别`的，也就是`所有 Index，Shard 都共用`的。

IndicesQueryCache 实现了 QueryCache 接口，这个接口是Lucene提供的用户缓存查询结果的接口

#### Lucene QueryCache，QueryCachingPolicy ，缓存类，缓存策略 ####

public interface QueryCache { 
    
      /** * Return a wrapper around the provided <code>weight</code> that will cache * matching docs per-segment accordingly to the given <code>policy</code>. * NOTE: The returned weight will only be equivalent if scores are not needed. * @see Collector#scoreMode() */
      Weight doCache(Weight weight, QueryCachingPolicy policy);
    
    }
    
    public interface QueryCachingPolicy { 
    
      /** Callback that is called every time that a cached filter is used. * This is typically useful if the policy wants to track usage statistics * in order to make decisions. */
      void onUse(Query query);
    
      /** Whether the given {@link Query} is worth caching. * This method will be called by the {@link QueryCache} to know whether to * cache. It will first attempt to load a {@link DocIdSet} from the cache. * If it is not cached yet and this method returns <tt>true</tt> then a * cache entry will be generated. Otherwise an uncached scorer will be * returned. */
      boolean shouldCache(Query query) throws IOException;
    
    }

Lucene提供了 `QueryCache` 和 `QueryCachingPolicy` 两个接口来处理检索结果缓存，ES中用 `ElasticsearchLRUQueryCache` 作为 QueryCache 的实现，默认使用Lucene提供的 `UsageTrackingQueryCachingPolicy` 作为QueryCachingPolicy的实现，处理是否缓存以及检索时的操作。

#### Lucene IndexSearcher 实例化 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NTI5OTE3_size_16_color_FFFFFF_t_70_pic_center]  
Elasticsearch在构建Lucene的IndexSearcher时，指定了 `QueryCache` 和 `QueryCachingPolicy`，QueryCache使用了装饰模式，`OptOutQueryCache` 是Security相关的操作，`IndicesQueryCache` 就是上面代码实例化的，内部持有 `ElasticsearchLRUQueryCache` 。

### IndexSearcher 检索 ###

#### Create Cache Weight ####

public class org.apache.lucene.search.IndexSearcher { 
    	// 查询缓存
        private QueryCache queryCache ;
        // 查询缓存策略
        private QueryCachingPolicy queryCachingPolicy ;
    
    	public Weight createWeight(Query query, boolean needsScores, float boost) throws IOException { 
            final QueryCache queryCache = this.queryCache;
            Weight weight = query.createWeight(this, needsScores, boost);
            // 只有不需要score的检索且缓存有配置 才能缓存结果
            if (needsScores == false && queryCache != null) { 
                // 查询缓存包装Weight, 返回CachingWrapperWeight
                weight = queryCache.doCache(weight, queryCachingPolicy);
            }
            return weight;
        }
    }

想要缓存检索结果，需要满足两个前提

1.  有设置`QueryCache`，这点在ES实例化`IndexSearcher`时已经制定了
2.  当前检索不需要实时的计算评分，不需要评分就代表着不需要计算`idf`，这样document的增删改就不会对当前缓存结果有影响。

其实从这里我们就能推断出`缓存的结果是基于segment的`，`倒排索引不可变，也就会生成的segment不可变，这是缓存生效的前提`。当对当前segment的数据做修改和删除时，变更信息记录在 `.liv` 文件内，不会直接修改segment原始文件。

#### QueryCachingPolicy\#onUse(Query query)，前置处理Query类型 ####

public class UsageTrackingQueryCachingPolicy implements QueryCachingPolicy { 
    
    	public void onUse(Query query) { 
            // 如果从不缓存
            if (shouldNeverCache(query)) { 
                return;
            }
            // 记录query的hash值
            int hashCode = query.hashCode();
            synchronized (this) { 
                // 添加最近使用的Filter， 通过记录hash值形式
                recentlyUsedFilters.add(hashCode);
            }
        }
    
        private static boolean shouldNeverCache(Query query) { 
           // 如果是通过Term查询的,不缓存, 因为查询效率很高,没有缓存的必要
           if (query instanceof TermQuery) { 
               // We do not bother caching term queries since they are already plenty fast.
               return true;
           }
           // 如果查询所有DOC, 结果集比Bit Set 遍历更快
           if (query instanceof MatchAllDocsQuery) { 
               // MatchAllDocsQuery has an iterator that is faster than what a bit set could do.
               return true;
           }
           // 不查询数据的query
           if (query instanceof MatchNoDocsQuery) { 
               return true;
           }
           if (query instanceof BooleanQuery) { 
               BooleanQuery bq = (BooleanQuery)query;
               // 没有查询条件从句，变相等于Match All
               if (bq.clauses().isEmpty()) { 
                   return true;
               }
           }
           if (query instanceof DisjunctionMaxQuery) { 
               DisjunctionMaxQuery dmq = (DisjunctionMaxQuery)query;
               if (dmq.getDisjuncts().isEmpty()) { 
                   return true;
               }
           }
           return false;
       }
    
    }

QueryCache对Query的类型有要求，当满足条件时，通过`记录hashcode的形式来记录query的使用次数`，这对后续是否缓存有影响。

#### LRUQueryCache\#shouldCache，前置处理segment的数据量 ####

public class LRUQueryCache implements QueryCache, Accountable { 
    
    	private boolean shouldCache(LeafReaderContext context) throws IOException { 
            // 缓存数据不能超过RAM上限
            return cacheEntryHasReasonableWorstCaseSize(ReaderUtil.getTopLevelContext(context).reader().maxDoc())
                // 缓存的segment的doc个数超过10000, 且占全部doc个数比例超过3%
                && leavesToCache.test(context);
        }
    
    	private boolean cacheEntryHasReasonableWorstCaseSize(int maxDoc) { 
            // The worst-case (dense) is a bit set which needs one bit per document
            // 最坏情况下, 使用BitSet缓存docID, 每个docID占用一个比特, 8个docID占用1字节
            final long worstCaseRamUsage = maxDoc / 8;
            // Cache指定的内存上限, 默认10%堆内存
            final long totalRamAvailable = maxRamBytesUsed;
            // 当前segment的docID的数据量不能超过Cache上限的20%,也就是堆内存的2%
            // 因为Cache是Node级别的, 如果一次性缓存数据太多, 会淘汰大量的已缓存数据, 类似于Mysql的全表扫描对缓存池的影响
            return worstCaseRamUsage * 5 < totalRamAvailable;
        }
    
    	public boolean test(LeafReaderContext context) { 
            // 当前segment的最大doc序号, 也就是当前segment有多少个document
            final int maxDoc = context.reader().maxDoc();
            // 如果当前segment的doc个数 < 10000, 那么久不缓存了
            if (maxDoc < minSize) { 
                return false;
            }
            final IndexReaderContext topLevelContext = ReaderUtil.getTopLevelContext(context);
            // 当前segment的doc个数 / 所有segment的doc个数
            final float sizeRatio = (float)context.reader().maxDoc() / topLevelContext.reader().maxDoc();
            // 当前segment的doc个数全部doc个数的比例超过 3%
            return sizeRatio >= minSizeRatio;
        }
    
    }

QueryCache还对`Segment的数据量有要求，太多和太少的都不缓存`。数据量太多，可能会触发热点缓存被大量淘汰，导致后续需要重新查询；太少则缓存没有意义，重新查询一样很快。

#### QueryCachingPolicy\#shouldCache，当前检索结果是否该被缓存 ####

public class UsageTrackingQueryCachingPolicy implements QueryCachingPolicy { 
    
    	public boolean shouldCache(Query query) throws IOException { 
    		// 上文代码中已经看过
            if (shouldNeverCache(query)) { 
                return false;
            }
            // 之前使用此query的次数, 通过query的hashcode来赋值取值, 在onUser代码里记录了query的hashcode
            final int frequency = frequency(query);
            // 2次或者5次
            final int minFrequency = minFrequencyToCache(query);
            // 最近使用的次数 >= 最小被缓存的次数
            return frequency >= minFrequency;
        }
    
    	int frequency(Query query) { 
            int hashCode = query.hashCode();
            synchronized (this) { 
            	// 取hashcode被记录的次数, 就是Query次数
                return recentlyUsedFilters.frequency(hashCode);
            }
        }
    
    	protected int minFrequencyToCache(Query query) { 
            // 如果是耗费比较大的查询请求
            if (isCostly(query)) { 
                return 2;
            } else { 
                // default: cache after the filter has been seen 5 times
                int minFrequency = 5;
                if (query instanceof BooleanQuery || query instanceof DisjunctionMaxQuery) { 
                    minFrequency--;
                }
                return minFrequency;
            }
        }
    }

当QueryCache未命中时，需要判断是否要缓存这次检索结果，如果符合要求，则执行检索，再缓存。

#### LRUQueryCache\#buildCache， 构建缓存数据 ####

public class LRUQueryCache implements QueryCache, Accountable { 
    
    	/** * Default cache implementation: uses {@link RoaringDocIdSet} for sets that have a density < 1% and a {@link BitDocIdSet} over a {@link FixedBitSet} * otherwise. */
        protected DocIdSet cacheImpl(BulkScorer scorer, int maxDoc) throws IOException { 
            // scorer.cost() 得到的是在当前segment里命中的doc数量
            if (scorer.cost() * 100 >= maxDoc) { 
                // FixedBitSet is faster for dense sets and will enable the random-access
                // optimization in ConjunctionDISI
                // 当命中doc数量占当前segment的比例超过1%时, 用BitSet存储
                return cacheIntoBitSet(scorer, maxDoc);
            } else { 
                // 用DocId数组存储
                return cacheIntoRoaringDocIdSet(scorer, maxDoc);
            }
        }
    
    	// 使用BitSet结构缓存数据
    	private static DocIdSet cacheIntoBitSet(BulkScorer scorer, int maxDoc) throws IOException { 
            final FixedBitSet bitSet = new FixedBitSet(maxDoc);
            long cost[] = new long[1];
            scorer.score(new LeafCollector() { 
    
                @Override
                public void setScorer(Scorer scorer) throws IOException { }
    
                @Override
                public void collect(int doc) throws IOException { 
                    cost[0]++;
                    // 缓存数据仅docId
                    bitSet.set(doc);
                }
    
            }, null);
            return new BitDocIdSet(bitSet, cost[0]);
        }
    
    }

构建缓存数据时，当`命中的doc数量超过当前segment的1%时`，使用`BitSet`结构存储，因为其占用空间小，每个docId仅占用`1bit`；当`小于1%`时，使用`long数组`形式缓存docId，因为这样解析快，数据不大时很方便

经过上述步骤就能缓存检索的DocIdSet，下次重复检索时就能实现复用。

### 总结 ###

`检索不需要评分是Cache生效的前提`，Must，Should都要算分，所以都不能使用Cache。

`倒排索引不可变是Filter Cache机制的基础，Cache是基于Segment的`，当segment经过merge时，对应的segment的Cache会失效，比如不断的Merge，被merge的Cache失效，但其他的segment还是可用的。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NTI5OTE3_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/4b8ed37dc12144829321b5ac3dec1083.png