ElasticsearchElasticsearch 查询过程中的 pre-filter 原理

Posted 2021-08-01 九师兄

1.概述

转载：添Elasticsearch 查询过程中的 pre-filter 原理

大家都知道在对索引执行查询的时候，需要在所有的分片上执行查询，因为无法知道被查询的关键词位于哪个分片，对于全文查询来说诚然如此，然而对于时序型的索引，当你从 my_index-* 中执行 now-3d 的范围查询时，可能很多分片上都不存在被查询的数据范围，因此 es 从 v5.6 开始引入了 pre-filter 机制：对于 Date 类型的 Range 查询，在对分片执行搜索之前，先检查一下分片是否包括被查询的数据范围，如果查询的范围与分片持有的数据没有交集，就跳过该分片。

分布式搜索过程原先由两个阶段执行：查询阶段和取回阶段，在引入了 pre-filter 之后，分布式搜索过程变成了三个阶段：预过滤阶段（pre-filter）、查询阶段和取回阶段。pre-filter 在查询阶段之前执行。

协调节点收到客户端的查询请求后，向本次搜索涉及到的全部分片发送RPC 请求：

indices:data/read/search[can_match]

这次 RPC 请求以 shard 为单位并行发送，没有并发限制。待查询的 shard 有多少个，就并发发送多少个 RPC。然后等待全部 RPC 返回响应。

tips

此时发送的 RPC 请求没有超时限制。事实上，_search 请求的 timeout参数仅在整个分布式搜索的 query 阶段进行检查，并且不包括 PRC 层面，他只在数据节点收到协调节点发来的 RPC 后开始计时，检查 query 过程是否超时。fetch 阶段的 RPC，以及数据节点对 fetch 请求的处理均没有超时检查。

节点收到请求后，判断请求的范围和待查询的分片是否存在交集，返回是或否，然后协调节点跳过不存在交集的分片，向其他分片发送下一阶段（查询阶段）的请求。

本次查询跳过了多少分片可以通过查询结果中的 skipped 字段看到，如：

"_shards":{
	"total": 130,
	"successful": 130,
	"skipped": 129,
	"failed": 0
}

同时也来看一下手册对 skipped 字段的解释：

skipped

(Integer) Number of shards that skipped the request because a lightweight check helped realize that no documents could possibly match on this shard. This typically happens when a search request includes a range filter and the shard only has values that fall outside of that range.

2.什么情况下会执行 pre-filter

pre-filter 并不会在所有查询过程中执行，在 v7.4中，需要同时满足以下条件，才会执行 pre-filter ：

• 待查询的分片数大于 128（pre_filter_shard_size指定）
• 聚合请求不要求访问所有 doc。即非 Global Aggregation 或 “min_doc_count” 不为0

另外，非 Date 类型的数值查询虽然也会走 pre-filter流程，但内部不会去判断范围，虽然协调节点也会发送 can_match 的 RPC，但数据节点的响应会在 MappedFieldType#isFieldWithinQuery 中直接返回相交，所以没有分片会被 skip，未来这方面可能会有扩展。

3. pre-filter 实现原理

数据节点判断某个 Range 查询与分片是否存在交集，依赖于 Lucene 的一个重要特性：PointValues 。在早期的版本中，数值类型在 Lucene 中被转换成字符串存入倒排索引，但是由于范围查询效率比较低，从 Lucene 6.0开始，对于数值类型使用 BKD-Tree 来建立索引，内部实现为 PointValues。PointValues原本用于地理位置场景，但它在多维、一维数值查询上的表现也很出色，因此原先的数值字段(IntField，LongField，FloatField，DoubleField)被替换为（IntPoint，LongPoint，FloatPoint，DoublePoint）

关于 BKD-Tree 的性质请参阅其他资料，暂且只需要知道 Lucene为每个字段单独建立索引，对于数值字段生成 BKD-Tree，一个新的 segment 生成时会产生一个新的.dim和.dii文件。最重要的，可以获取到这个 segment 中数值字段的最大值和最小值，为 pre-filter 提供了基础。当 segment 被 reader 打开的时候，Lucene 内部的 BKDReader 会将最大值和最小值读取出来保存到类成员变量，因此每个 segment 中，每个数值字段的最大最小值都是常驻 JVM 内存的。

既然每个 segment 记录了数值字段的取值范围，获取shard 级别的范围就轻而易举：PointValues.getMaxPackedValue()，PointValues.getMinPackedValue(),函数遍历全部的 segment 分别计算最大值和最小值，然后根据查询条件判断是否存在交集，在 DateFieldMapper.DateFieldType#isFieldWithinQuery 函数中：

private Relation isFieldWithinRange(IndexReader reader, long fromInclusive, long toInclusive) throws IOException {
     if (PointValues.size(reader, name()) == 0) {
         // no points, so nothing matches
         return Relation.DISJOINT;
     }

     long minValue = LongPoint.decodeDimension(PointValues.getMinPackedValue(reader, name()), 0);
     long maxValue = LongPoint.decodeDimension(PointValues.getMaxPackedValue(reader, name()), 0);

     if (minValue >= fromInclusive && maxValue <= toInclusive) {
         return Relation.WITHIN;
     } else if (maxValue < fromInclusive || minValue > toInclusive) {
         return Relation.DISJOINT;
     } else {
         return Relation.INTERSECTS;
     }
 }

既然数值类型都可以获取分片级别的范围，为什么 pre-filter 只在 Date 类型的Range 查询里实现了，而其他的数值类型的 Range 查询不会走 pre-filter 流程？原因也非常简单，只有 Date 类型的数值确定是递增的，其他数值类型未必。对于非递增的数值字段，其数据会散布到 my_index-* 的每个分片上，因此 pre-filter 也就没有必要了。如果你有另外一个递增的数值字段，目前也没有配置的方式来使用 pre-filter。

题外话：BKD-Tree 的每个节点都记录了节点自己的maxPackedValue、minPackedValue

Lucenene 内部查询的也会按照 segment file 级别跳过

现在我们忘掉 es，讨论数值类型查询在 Lucene 内部的实现。

HBase 的写入模型和 Lucene 类似，先写内存，然后刷盘生成 HFile，HFile 合并成大文件。 由于 HBase 使用时间戳作为数据版本号，因此每个 HFile 都记录了时间范围。因此查询的时候如果指定时间范围，就可以过滤掉大量的 HFile 不用查询。这么优秀的操作在 Lucene 中也必不可少。

在一个 Lucene 索引中可能有很多 segment，Lucene遍历所有的 segment 进行处理，在对每个 segment 的 weight.bulkScorer过程中，BKDReader.intersect函数根据相交情况决定收集符合条件的 docid，如果查询条件和 segment 没有交集，就什么都不做。

因此当对数值类型查询的时候，不在范围的 segment 会直接跳过，Lucene 内部称为：CELL_OUTSIDE_QUERY

但是，段合并的时候目前还不会考虑按时间临近的方式进行合并，因此借鉴 HBase 的思想按照时间临近的段进行合并有助于降低数值类型的范围查询耗时。

4.思考

既然 Lucene 对数值类型有 segment 级别的skip，Elasticsearch 实现的分片层面的 pre-filter 还有必要存在吗？他可以让搜索延迟更低么？我们实际测试来说话。过程如下：

生产集群有 filebeat-* 索引，数据为 nginx 日志，大约8T，有180个shard，11227个 segment，分布在3个节点。

step1：我们对 date 字段执行一个不会命中的查询，让他走 pre-filter 流程：

POST filebeat-7.4.2-*/_search
{
  "size":0,
  "query": {
    "range": {
      "@timestamp": {
        "gte": "2021-01-01",
        "lte":"2022-01-01"
      }
    }
  }
}

返回结果摘要如下，整个过程执行了31ms：

{
  "took" : 31,
  "timed_out" : false,
  "_shards" : {
    "total" : 180,
    "successful" : 180,
    "skipped" : 179,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 0,
      "relation" : "eq"
    }
}

step2：现在加上 ?pre_filter_shard_size=1000 参数重新查询，其他条件不变，让查询过程不走 pre-filter，返回结果如下，整个过程执行了31ms，可见没什么区别：

{
  "took" : 23,
  "timed_out" : false,
  "_shards" : {
    "total" : 180,
    "successful" : 180,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 0,
      "relation" : "eq"
    }
  }
}

step3：最后我们对 long 字段执行 range 查询，这样也不走 pre-filter 流程：

POST filebeat-7.4.2-*/_search?size=0
{
  "query": {
    "range": {
      "nginx.bytes.body_sent": {
        "gte": -2,
        "lte":-1
      }
    }
  }
}

这次查询执行了50ms，还是在一个数据级。

{
  "took" : 50,
  "timed_out" : false,
  "_shards" : {
    "total" : 180,
    "successful" : 180,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 0,
      "relation" : "eq"
    }
  }
}

因此 pre-filter并不会降低查询延迟，在和官方聊过之后，他们的想法是 pre-filter 最主要的作用不是降低查询延迟，而是 pre-filter 阶段可以不占用search theadpool，减少了这个线程池的占用情况。个人感觉这个收益并不大。不过未来会在这个阶段做更多的查询优化， 例如7.6中放出的 #49092,#48681

特别感谢：陆徐刚@蚂蚁
基于：Elasticsearch 7.4 & 7.6

参考

https://github.com/elastic/elasticsearch/pull/25658
https://www.amazingkoala.com.cn/Lucene/Search/2020/0427/135.html
https://lucene.apache.org/core/6_2_1/core/org/apache/lucene/index/PointValues.html
https://www.jianshu.com/p/39eb0d66d082
https://cloud.tencent.com/developer/article/1366835
https://www.elastic.co/guide/en/elasticsearch/reference/current/release-notes-7.6.0.html#