Elasitcsearch 底层系列 Lucene 内核解析之 Doc Value
原创Elasitcsearch 底层系列 Lucene 内核解析之 Doc Value
原创
背景
Elasticsearch 支持行存和列存，行存用于以文档为单位顺序存储多个文档的原始内容，在 Elasitcsearch 底层系列 Lucene 内核解析之 Stored Fields 文章中介绍了行存的细节。列存则以字段为单位顺序存储多个文档同一字段的内容，主要用于排序、聚合、范围查询等场景，新版本的 ES 绝大部分字段都会保存 doc value，可以显示指定关闭。今天我们就来剖析 ES 列存（doc value）的细节。代码解析基于 ES 6.3/Lucene 7.3 的版本。
我们在腾讯云提供了原生的ES服务（CES）及CTSDB时序数据库服务，欢迎各位交流底层技术。
Doc value 的官方介绍：https://www.elastic.co/guide/en/elasticsearch/reference/current/doc-values.html
本文主要分以下几个部分介绍：
- 基本框架
- 文件结构
- 写入流程
- 读取流程
- 合并流程
基本框架
进入各个流程之前，我们先来看一下 doc Value 相关的类结构。下图蓝色是 doc value 的写入部分主要框架，文档的写入入口在 DefaultIndexingChain，每一个 field 都有对应的 PerField 对象，包含 field info 以及相关的写入类。写入的时候根据字段类型，例如 Binary、Numeric、StoredNumeric、SortedSet 等选择对应的 Writer进行处理。各个 Writer 负责内存中的写入及数据结构整理压缩逻辑，Lucene70DocValuesConsumer 负责底层文件的写入。红色部分是读取框架，同样也是按照不同类型分别处理读取，Lucene70DocValuesProducer 负责文件读取解析。后面的写入及读取流程我们再来详细剖析。
文件结构
Doc value 的 lucene 文件主要是 dvd 和 dvm 后缀文件，dvd 文件为数据文件，保存各种值， dvm 文件为数据文件的索引文件，便于快速解析查找数据文件。dvd 文件一般都比较大，dvm 文件都很小，如下图所示：
我们先来总体看一下文件的内容结构，后面再结合代码详细分析内容的生成和读取过程。
dvd 和 dvm 都有如下公共的文件头信息：
dvm 索引文件，除头尾信息以外，中间的部分主要是顺序保存每个字段编码相关的元数据信息，以及切分 block 的信息。
dvd 数据文件，除头尾信息以外，中间的部分主要是顺序保存每个字段编码压缩后的内容：
dvd 等值及 Multiple block 的场景：
当字段不是数值类型，会保存 value 的 hash 映射，该字段会分三层依次保存，第一层是每个 value 的 hash 位置，第二层是每个 value 的原始值，第三层是原始值的索引项。其中第一层结构和上述结构一致，第二、三层 dvm、dvd 结构如下所示，前半部分为 terms，后半部分为 terms 索引信息：
接下来结合这些文件结构，我们来分析代码是如何产生和读取这些内容的。
写入流程
先来看如下示例数据：
{
"@timestamp":"2017-03-23T13:00:00",
"accept":36320,
"deny":4156,
"host":"server_2",
"response":2.4558210155,
"service":"app_3",
"total":40476
}mapping 自动生成，ES 将会产生如下类型的字段：
本次重点关注标红的 DocValue 对象。
PackedInts(long)
在正式进入 doc value 剖析之前，我们先来看一个数据类型：PackedInts。它是 doc value 数值存储压缩使用的主要类型。数值类型列存有很大的压缩空间，可以节省很多内存开销。这种压缩是基于数据运算或者类型压缩实现的。
例如，假设某个列的值全是一样的（例如内置的 _version, _primary_term 字段，极有可能全一样），此时 PackedInt 可以简单的用一个整型对象存一个值即可。假设某个列的数值最大存储只需要 10 个 bit，我们直接用 short 存储会浪费6个 bit，内存浪费接近一半。
Lucene 中实现的 PackInts 对象会将内存划分为逻辑上的多个 block，每个 block 一定是8位内存对齐的，最常用的就是直接利用一个 long 对象作为一个 block，充分利用每个类型的每一个 bit，避免浪费。假设我们每个列的 value 用10个 bit 就可以存储，用 long 对象来储存多个 value 如下所示：
注意 value n会跨两个 block（long） 对象。
写入调用链时序
写入流程分为内存写入流程和刷新流程，以下是写入调用链时序：
入口在 DefaultIndexingChian，内存写入主要在各类型 DocValuesWriter中，flush 落盘主要在 Lucene70DocValuesConsumer中。接下来我们分别分析内存写入和刷新流程。
内存写入流程
在前面我们讲 Stored Fields 的时候，有提到 Lucene 的 index 动作是在 DefaultIndexingChain 类里面完成的，今天我们直接跳到对应的 doc value 处理的逻辑：
DefaultIndexingChain.processDocument()
DocValuesType dvType = fieldType.docValuesType();
if (dvType == null) {
throw new NullPointerException("docValuesType must not be null (field: \"" + fieldName + "\")");
}
if (dvType != DocValuesType.NONE) {
if (fp == null) {
fp = getOrAddField(fieldName, fieldType, false);
}
indexDocValue(fp, dvType, field); // 内存中处理每个 field 的 doc value
}这里的 indexDocValue 函数完成了 doc value 的保存逻辑。进到该函数里面，会对每个字段的 doc value 类型做分类处理，如下的每个分支就对应着上述各字段类型的写入操作。每个字段都会对应一个 DocValueWriter。
DefaultIndexingChain.indexDocValue
switch(dvType) {
case NUMERIC:
if (fp.docValuesWriter == null) {
fp.docValuesWriter = new NumericDocValuesWriter(fp.fieldInfo, bytesUsed);
}
if (field.numericValue() == null) {
throw new IllegalArgumentException("field=\"" + fp.fieldInfo.name + "\": null value not allowed");
}
((NumericDocValuesWriter) fp.docValuesWriter).addValue(docID, field.numericValue().longValue());
break;
case BINARY:
if (fp.docValuesWriter == null) {
fp.docValuesWriter = new BinaryDocValuesWriter(fp.fieldInfo, bytesUsed);
}
((BinaryDocValuesWriter) fp.docValuesWriter).addValue(docID, field.binaryValue());
break;
case SORTED:
if (fp.docValuesWriter == null) {
fp.docValuesWriter = new SortedDocValuesWriter(fp.fieldInfo, bytesUsed);
}
((SortedDocValuesWriter) fp.docValuesWriter).addValue(docID, field.binaryValue());
break;
case SORTED_NUMERIC:
if (fp.docValuesWriter == null) {
fp.docValuesWriter = new SortedNumericDocValuesWriter(fp.fieldInfo, bytesUsed);
}
((SortedNumericDocValuesWriter) fp.docValuesWriter).addValue(docID, field.numericValue().longValue());
break;
case SORTED_SET:
if (fp.docValuesWriter == null) {
fp.docValuesWriter = new SortedSetDocValuesWriter(fp.fieldInfo, bytesUsed);
}
((SortedSetDocValuesWriter) fp.docValuesWriter).addValue(docID, field.binaryValue());
break;
default:
throw new AssertionError("unrecognized DocValues.Type: " + dvType);
}最常使用的类型是 SortedNumericDocValuesWriter 和 SortedSetDocValuesWriter ，因为 doc value 主要用在聚合排序等操作上，上述两种类型的 writer 分别对应了数值类型和字符类型的 doc value 排序写操作。这里的 Sorted 关键字排序是指“同一个文档中该字段的多个 value （数组）之间进行排序“，不是指“多个文档按照该字段进行排序”。多个文档之间的排序由 index level sorting 决定。接下来我们重点分析这两种数据类型的写入。
SortedNumericDocValuesWriter
数值类型 doc value 的写操作。从前面的 case 分支中可以看到，每一个字段的 DocValueWriter 会在第一次进来的时候被初始化，一个 field 对应一个 docValuesWriter：
DefaultIndexingChain.indexDocValue
case SORTED_NUMERIC:
if (fp.docValuesWriter == null) {
fp.docValuesWriter = new SortedNumericDocValuesWriter(fp.fieldInfo, bytesUsed);
}
((SortedNumericDocValuesWriter) fp.docValuesWriter).addValue(docID, field.numericValue().longValue());
break;SortedNumericDocValuesWriter 对象的初始化逻辑：
SortedNumericDocValuesWriter.java
public SortedNumericDocValuesWriter(FieldInfo fieldInfo, Counter iwBytesUsed) {
this.fieldInfo = fieldInfo;
this.iwBytesUsed = iwBytesUsed;
// 保存 value 对象，页满时 pack，一页最多1024个 value ，pack 后放到 values 对象中，在 flush 的时候会通过 build 函数取出
pending = PackedLongValues.deltaPackedBuilder(PackedInts.COMPACT);
// 保存 每个文档中当前字段 value 的数量，单个 field 每个文档可能存在多个 doc value
pendingCounts = PackedLongValues.deltaPackedBuilder(PackedInts.COMPACT);
// 保存 docId，这里的 docId 只记录最大值，取的时候顺序+1取
docsWithField = new DocsWithFieldSet();
bytesUsed = pending.ramBytesUsed() + pendingCounts.ramBytesUsed() + docsWithField.ramBytesUsed() + RamUsageEstimator.sizeOf(currentValues);
iwBytesUsed.addAndGet(bytesUsed);
}Number 类型的载体对象都是 PackedLongValues, 该对象的构造过程：
public static PackedLongValues.Builder deltaPackedBuilder(float acceptableOverheadRatio) {
// 默认页大小是 1024
// 这里 acceptableOverheadRatio 取值默认为0，表示最佳压缩模式，充分利用每个 bit
return deltaPackedBuilder(DEFAULT_PAGE_SIZE, acceptableOverheadRatio);
}在前面有看到传入构造的参数是：PackedInts.COMPACT，表示最佳压缩，不浪费一个 bit。这里Packed等级有四种，不同的等级表示可以允许多少内存的浪费率，浪费的空间会自动内存补齐。浪费多效率高，浪费少效率低，这里是时间换空间的概念。
/**
* At most 700% memory overhead, always select a direct implementation.
*/
public static final float FASTEST = 7f;
/**
* At most 50% memory overhead, always select a reasonably fast implementation.
*/
public static final float FAST = 0.5f;
/**
* At most 25% memory overhead.
*/
public static final float DEFAULT = 0.25f;
/**
* No memory overhead at all, but the returned implementation may be slow.
*/
public static final float COMPACT = 0f;相关的初始化工作只在字段第一次处理 doc value 的时候进行，初始化完成之后就进入添加值阶段。在上述 indexDocValue 函数中的 case 语句中，根据每个类型进来调用对应 writer 的 addValue 方法保存 doc value。addValue 的逻辑都差不多，以 SortedNumericDocValuesWriter 为例如下所示：
SortedNumericDocValuesWriter.java
public void addValue(int docID, long value) {
assert docID >= currentDoc;
if (docID != currentDoc) { // 新进来 doc 先结束上次的 doc
finishCurrentDoc();
currentDoc = docID;
}
addOneValue(value); // 添加值
updateBytesUsed();
}addOneValue 只是简单的将值添加到一个自扩容的 long 型数组中：
private void addOneValue(long value) {
if (currentUpto == currentValues.length) {
// 空间不够就扩容
currentValues = ArrayUtil.grow(currentValues, currentValues.length+1);
}
currentValues[currentUpto] = value; //long currentValues[]
currentUpto++; // 更新值下标
}finishCurrentDoc 的逻辑，主要是将上述添加的数组保存到 pending 中，pending 是一个 PackedLongValues 的 builder 对象，其内部会判断是否达到 pack 的条件，达到就进行 pack。
private void finishCurrentDoc() {
if (currentDoc == -1) {
return;
}
// 这里是对同一个 doc 中的该字段的多个 doc value 进行内部排序，SortedNumeric 的 Sort 就在这里体现
Arrays.sort(currentValues, 0, currentUpto);
for (int i = 0; i < currentUpto; i++) {
pending.add(currentValues[i]); // PackedLongValues
}
// record the number of values for this doc
pendingCounts.add(currentUpto); // 当前 doc 中该字段的 doc value 数量，一般情况是 1
currentUpto = 0;
docsWithField.add(currentDoc); // 保存当前 doc id
}
接下来我们看一下上述 pending.add 函数的详细实现 ：
PackedLongValues.java
/** Add a new element to this builder. */
public Builder add(long l) {
if (pending == null) {
throw new IllegalStateException("Cannot be reused after build()");
}
if (pendingOff == pending.length) { // 达到 1024 个对象，pack 一次
// check size
if (values.length == valuesOff) { // values 保存 pack 后的对象，默认长度 16，不够自动扩容
final int newLength = ArrayUtil.oversize(valuesOff + 1, 8);
grow(newLength);
}
pack(); // 压缩处理，处理 pending 中的内容，pack 完毕之后 pendingOff 会置零
}
pending[pendingOff++] = l; // 简单的添加对象到 pending 中保存，pending 最大 1024
size += 1;
return this;
}
接着看 pack 的具体逻辑，它是实现压缩的主要函数：
PackedInts.java
void pack(long[] values, int numValues, int block, float acceptableOverheadRatio) {
assert numValues > 0;
// compute max delta
long minValue = values[0];
long maxValue = values[0];
for (int i = 1; i < numValues; ++i) {
minValue = Math.min(minValue, values[i]);
maxValue = Math.max(maxValue, values[i]);
}
// build a new packed reader
if (minValue == 0 && maxValue == 0) {
// 数值类的对象进来后先求最小最大值，如果全部都是相同的值，比如 version 全为1，primary term 全为 0 等场景，直接保存一个值即可
this.values[block] = new PackedInts.NullReader(numValues);
} else {
// 计算最大值所需的 bit 数量
final int bitsRequired = minValue < 0 ? 64 : PackedInts.bitsRequired(maxValue);
// 根据大小分配一个合适可变对象，后面详述
final PackedInts.Mutable mutable = PackedInts.getMutable(numValues, bitsRequired, acceptableOverheadRatio);
for (int i = 0; i < numValues; ) {
i += mutable.set(i, values, i, numValues - i); // 将 values 对象 pack 到 mutable 对象中，后面详述
}
this.values[block] = mutable; // pack 后的对象保存到 values 数组中，后面会写入磁盘
}
}PackedInts.getMutable 的实现逻辑：
PackedInts.java
public static Mutable getMutable(int valueCount,
int bitsPerValue, float acceptableOverheadRatio) {
// 根据配置的压缩比的类型（COMPACT、FASTEST等）计算压缩时采取的 bitsPerValue 数量，
// 以及是否有必要压缩，返回的 formatAndBits.format 参数一般情况取值为 Format.PACKED 表示压缩。
final FormatAndBits formatAndBits = fastestFormatAndBits(valueCount, bitsPerValue, acceptableOverheadRatio);
// 根据类型和值的 bit 数量选取合适的 Pakced 对象，如果所需 bit 数刚好是 8 的整数倍，
// 则直接用 Direct8、Direct16、Direct32、Direct64 来存储，否则会用 Packed64 对象（long）存储。
return getMutable(valueCount, formatAndBits.bitsPerValue, formatAndBits.format);
}我们拿 Packed64 为例讲一下上述 pack 中的 set 逻辑：
Packed64.java
@Override
public int set(int index, long[] arr, int off, int len) {
// of 函数里面的重点是根据 bitsPerValue 即 doc value 中最大的值所需的 bit 数量，
// 来确定写的 encode 对象，例如 BulkOperationPacked10 表示最大的需要 10 个 bit
...
final PackedInts.Encoder encoder = BulkOperation.of(PackedInts.Format.PACKED, bitsPerValue);
...
// 编码的逻辑就在对应的 encode 函数中，后面详述
encoder.encode(arr, off, blocks, blockIndex, iterations);
...
}BulkOperationPacked10（最大到24）对象构造函数调用 BulkOperationPacked 传递对应的 bit 数：
public BulkOperationPacked10() {
super(10); // 调用父类 BulkOperationPacked 构造函数，下面详述
}BulkOperationPacked 的构造函数逻辑：
public BulkOperationPacked(int bitsPerValue) {
this.bitsPerValue = bitsPerValue; // value 需要的最大 bit 数
assert bitsPerValue > 0 && bitsPerValue <= 64;
int blocks = bitsPerValue;
// 这里算需要多少个 block 即 long 对象能够完整的保存 n 个 value （简单的判断能被2整除就行）
// 例如 bitsPerValue 是10，则至少需要5个 long 对象才不需要跨 long 保存 (5*32=320 才刚好被10整除，能保存32个 value 对象)
while ((blocks & 1) == 0) {
blocks >>>= 1;
}
this.longBlockCount = blocks;
this.longValueCount = 64 * longBlockCount / bitsPerValue; // 根据算好的 long block 数量计算能保存的 value 数量
...
}上面讲的 BulkOperationPacked10 是继承至 BulkOperationPacked 类，主要的压缩编码逻辑都在 BulkOperationPacked 类中的 encode 函数中实现，将多个 value 保存到连续的 long 对象中，这个函数是整个压缩编码的核心：
BulkOperationPacked.java
/**
* values: 被压缩的数组对象
* valuesOffset: 被压缩数组对象的偏移（index），顺序加一取 values
* blocks: 压缩此数组对象所需的 long 对象数组，目标输出对象
* blcoksOffset：block 对象的 index
* iterations：longValueCount * iterations = 总的 values 的长度
*
* 示例如下：
* 假设 values 数组有1024个元素，bitsPerValue = 10（即最大的元素需要10个 bit 存储），
* 那么共需要 1024*10=10240 个 bit，10240/8=1280 个 byte，1280/8=160 个 long， blocks 的长度就是160
*/
@Override
public void encode(long[] values, int valuesOffset, long[] blocks,
int blocksOffset, int iterations) {
long nextBlock = 0;
int bitsLeft = 64;
// 遍历待压缩的 values 对象
for (int i = 0; i < longValueCount * iterations; ++i) {
bitsLeft -= bitsPerValue; // 每个对象都占用 bitsPerValue 位
if (bitsLeft > 0) { // 直到一个 long 对象分配完毕
nextBlock |= values[valuesOffset++] << bitsLeft; // 移位操作将多个 values 压缩成一个 long
} else if (bitsLeft == 0) { // 刚好用完
nextBlock |= values[valuesOffset++];
blocks[blocksOffset++] = nextBlock;
nextBlock = 0;
bitsLeft = 64;
} else { // bitsLeft < 0 某个 values 对象跨两个 long
nextBlock |= values[valuesOffset] >>> -bitsLeft;
blocks[blocksOffset++] = nextBlock;
nextBlock = (values[valuesOffset++] & ((1L << -bitsLeft) - 1)) << (64 + bitsLeft);
bitsLeft += 64;
}
}
}上面就是SortedNumericDocValuesWriter写入的过程，经过 PackedInt 压缩编码之后，数据会以相对节省的形式存放在内存中。接下来我们看可能看字符类型的写入流程。
SortedSetDocValuesWriter
该对象主要处理字符类型的 doc value 写逻辑。其内部会用一个 BytesRefHash 对象保存字符的 byte 数组，以及对应的 hash 位置（termId），termId 会像上述 NumericDocValue 一样采用 PackedInts 压缩。BytesRefHash 内部有一个 ByteBlockPool，其成员变量 byte[] buffer 中保存了字符 byte 数组。我们看一下 SortedSetDocValuesWriter 的添加值的逻辑：
SortedSetDocValuesWriter.java
private void addOneValue(BytesRef value) {
int termID = hash.add(value); // BytesRefHash 对象，add 动作添加 byte 数组并计算对应的 hash 值并返回
......
currentValues[currentUpto] = termID; // 添加字符对象的 hash 值
currentUpto++;
}以上就是内存写入流程，采用 PackedInts 类型，可以最大程度的节省内存。内存写入后，doc value 对象都是以该类型保存在内存中，后面的刷新流程会将内存中的 doc value 反编码解压，之后以紧凑型 byte 数组写入 segment 文件（dvd）。
刷新流程
刷新流程的入口在 DefaultIndexingChain.writeDocValues 中。writeDocValues 只是 DefaultIndexingChain.flush 的一个步骤，flush 函数包含了其它类型例如 stored fields，norms，point 等类型的刷新逻辑。DocValue刷新的时候会将各个字段顺序刷到 dvd、dvm 文件。下面是 writeDocValues 的详细分析：
DefaultIndexingChain.java
/** Writes all buffered doc values (called from {@link #flush}). */
private void writeDocValues(SegmentWriteState state, Sorter.DocMap sortMap) throws IOException {
int maxDoc = state.segmentInfo.maxDoc(); // 这个 segment 当前在内存中的文档数
DocValuesConsumer dvConsumer = null;
boolean success = false;
try {
for (int i=0;i<fieldHash.length;i++) {
// 遍历每一个 field 逐个顺序刷盘，PerField 里面保存的 fieldInfo，fieldInfo 包含了字段名、类型等基本信息
PerField perField = fieldHash[i];
while (perField != null) {
if (perField.docValuesWriter != null) { // 如果是 doc value 类型的，则之前肯定用 docValuesWriter（例如 NumericDocValuesWriter） 写过数据进内存
if (perField.fieldInfo.getDocValuesType() == DocValuesType.NONE) {
// BUG
throw new AssertionError("segment=" + state.segmentInfo + ": field=\"" + perField.fieldInfo.name + "\" has no docValues but wrote them");
}
if (dvConsumer == null) {
// lazy init
DocValuesFormat fmt = state.segmentInfo.getCodec().docValuesFormat();
// 初始化 Lucene70DocValuesConsumer ，调用 Lucene70DocValuesConsumer 的构造函数创建（若未创建）dvd，dvm文件，并写入 header 信息
dvConsumer = fmt.fieldsConsumer(state);
}
if (finishedDocValues.contains(perField.fieldInfo.name) == false) {
perField.docValuesWriter.finish(maxDoc); // 调用 DocValueWriter 的finish，对未完成的值做一轮 pack
}
perField.docValuesWriter.flush(state, sortMap, dvConsumer); // 主要的刷新调用逻辑，后面详细分析
perField.docValuesWriter = null;
} else if (perField.fieldInfo.getDocValuesType() != DocValuesType.NONE) {
// BUG
throw new AssertionError("segment=" + state.segmentInfo + ": field=\"" + perField.fieldInfo.name + "\" has docValues but did not write them");
}
perField = perField.next;
}
}上面主要的 flush 函数是由各个类型的 DocValuesWriter 来实现的，常用的 writer 类型：
- NumericDocValuesWriter （数字类型）
- SortedNumericDocValuesWriter （多值内部排序的数值类型）
- SortedDocValuesWriter （排序的字符类型，保存原始值及 hash 位置）
- SortedSetDocValuesWriter （排序的字符数组类型，保存原始值及 hash 位置）
每种类型的 flush 函数的结构都是类似的，分为三部分：
- build 缓存在 pending 中的对象，生成 PackedLongValues。PackedLongValues 对象包含两个最主要的数组成员，一个是 mins，保存每个 pack 后对象的最小值（每个 value 会算差值）；另一个是 values，保存实际 pack 后的对象，例如 Packed64, DirectInt 等，取决于 doc value bit 使用数量。
- 根据索引排序字段顺序对 doc value 进行排序。
- 写处理好的 value 进 dvd 文件，同时写 dvm 索引文件。
以 SortedNumericDocValuesWriter 为例：
SortedNumericDocValuesWriter.java
@Override
public void flush(SegmentWriteState state, Sorter.DocMap sortMap, DocValuesConsumer dvConsumer) throws IOException {
// build 缓存在 pending 中的对象，生成 PackedLongValues
final PackedLongValues values;
final PackedLongValues valueCounts;
if (finalValues == null) {
values = pending.build();
valueCounts = pendingCounts.build();
} else {
values = finalValues;
valueCounts = finalValuesCount;
}
// 排序，这里的排序是 index sorting 指定的排序，会按照排序的字段传进来一个 sortMap，这个 sortMap 就是按照排序字段排好的 docId
final long[][] sorted;
if (sortMap != null) {
sorted = sortDocValues(state.segmentInfo.maxDoc(), sortMap,
new BufferedSortedNumericDocValues(values, valueCounts, docsWithField.iterator()));
} else {
sorted = null;
}
// 写 dvd dvm 文件，后面详细描述
dvConsumer.addSortedNumericField(fieldInfo,
new EmptyDocValuesProducer() {
@Override
public SortedNumericDocValues getSortedNumeric(FieldInfo fieldInfoIn) {
if (fieldInfoIn != fieldInfo) {
throw new IllegalArgumentException("wrong fieldInfo");
}
// 读取内存中缓存的 values
final SortedNumericDocValues buf =
new BufferedSortedNumericDocValues(values, valueCounts, docsWithField.iterator());
if (sorted == null) {
return buf;
} else {
return new SortingLeafReader.SortingSortedNumericDocValues(buf, sorted);
}
}
});
}上面读取内存缓存的 values 主要用到 BufferedSortedNumericDocValues 类，该类构造方法传入我们之前压缩的 values （Packed64, DirectInt等）。在构造函数中会对压缩的内容进行解压，主要调用 BulkOperationPacked10（例）decode 函数解压，解压逻辑是每次将一个 block（long）偏移10位计算对应的值放到 values 数组中。
接下来我们看看 dvConsumer.addSortedNumericField 的实现逻辑，该函数中主要的逻辑是调用 writeValues 函数实现的：
Lucene70DocValuesConsumer.java
private long[] writeValues(FieldInfo field, DocValuesProducer valuesProducer) throws IOException {
SortedNumericDocValues values = valuesProducer.getSortedNumeric(field);
int numDocsWithValue = 0;
MinMaxTracker minMax = new MinMaxTracker();
MinMaxTracker blockMinMax = new MinMaxTracker();
long gcd = 0;
Set<Long> uniqueValues = new HashSet<>();
// 下面这个 for 循环计算 segment 所有 value 的最小最大，以及每个 block 的最小最大，并记录最大公约数和唯一值，便于后面选择压缩策略
for (int doc = values.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = values.nextDoc()) {
for (int i = 0, count = values.docValueCount(); i < count; ++i) {
long v = values.nextValue();
if (gcd != 1) {
if (v < Long.MIN_VALUE / 2 || v > Long.MAX_VALUE / 2) {
// in that case v - minValue might overflow and make the GCD computation return
// wrong results. Since these extreme values are unlikely, we just discard
// GCD computation for them
gcd = 1;
} else if (minMax.numValues != 0) { // minValue needs to be set first
gcd = MathUtil.gcd(gcd, v - minMax.min);
}
}
minMax.update(v);
blockMinMax.update(v);
if (blockMinMax.numValues == NUMERIC_BLOCK_SIZE) {//达到一个 block size 的时候 reset 一下
blockMinMax.nextBlock();
}
// 记录不重复值的数量，如果小于 256 个，则稍后采用 unique 压缩方法，去掉不必要的重复值
if (uniqueValues != null
&& uniqueValues.add(v)
&& uniqueValues.size() > 256) {
uniqueValues = null;
}
}
numDocsWithValue++; //含有值的文档数量
}
minMax.finish();
blockMinMax.finish();
final long numValues = minMax.numValues; // 值的数量
long min = minMax.min;
final long max = minMax.max;
assert blockMinMax.spaceInBits <= minMax.spaceInBits;
if (numDocsWithValue == 0) {
// 包含值的文档数为0，即该 segment 中所有文档中都不包含该字段值
meta.writeLong(-2);
meta.writeLong(0L);
} else if (numDocsWithValue == maxDoc) {
// 满值的场景，segment 文档数量刚好和含有值的文档数量相等
meta.writeLong(-1);
meta.writeLong(0L);
} else {
// 稀疏场景，segment 中有部分文档不包含值，这里要用 bit set 来记录哪些文档包含值
long offset = data.getFilePointer();
meta.writeLong(offset);
values = valuesProducer.getSortedNumeric(field);
IndexedDISI.writeBitSet(values, data);
meta.writeLong(data.getFilePointer() - offset);
}
meta.writeLong(numValues); // 记录值的数量
final int numBitsPerValue;
boolean doBlocks = false;
Map<Long, Integer> encode = null;
if (min >= max) {
// 最小值和最大值相等的场景，meta 标记一下，稍后 data 直接写一个最小值即可
numBitsPerValue = 0;
meta.writeInt(-1);
} else {
if (uniqueValues != null
&& uniqueValues.size() > 1
&& DirectWriter.unsignedBitsRequired(uniqueValues.size() - 1) < DirectWriter.unsignedBitsRequired((max - min) / gcd)) {
// 唯一值的数量小于 256 的场景，这里会先在 meta 中直接记录排序后的不重复值，后面 data 中记录值的位置即可
numBitsPerValue = DirectWriter.unsignedBitsRequired(uniqueValues.size() - 1);
final Long[] sortedUniqueValues = uniqueValues.toArray(new Long[0]);
Arrays.sort(sortedUniqueValues);
meta.writeInt(sortedUniqueValues.length);
for (Long v : sortedUniqueValues) {
meta.writeLong(v);
}
encode = new HashMap<>();
for (int i = 0; i < sortedUniqueValues.length; ++i) {
encode.put(sortedUniqueValues[i], i); // encode 保存值的索引，用于在 data 中记录位置
}
min = 0;
gcd = 1;
} else {
uniqueValues = null;
// 这里检查每个 block 的使用空间加起来的大小和不划分 block 整体的使用空间大小，差别太大就划分 block
// we do blocks if that appears to save 10+% storage
doBlocks = minMax.spaceInBits > 0 && (double) blockMinMax.spaceInBits / minMax.spaceInBits <= 0.9;
if (doBlocks) {
numBitsPerValue = 0xFF;
meta.writeInt(-2 - NUMERIC_BLOCK_SHIFT); // 多 block 标记
} else {
numBitsPerValue = DirectWriter.unsignedBitsRequired((max - min) / gcd);
if (gcd == 1 && min > 0
&& DirectWriter.unsignedBitsRequired(max) == DirectWriter.unsignedBitsRequired(max - min)) {
min = 0; // 最小最大值差异太大，差值没法改善压缩，例如 1,3,9...45664545,53545465,46567677。如果都是很大的值则都减掉最小值可以起到压缩作用。
}
meta.writeInt(-1); // 单个 block 标记
}
}
}
meta.writeByte((byte) numBitsPerValue); // 记录每个值所需的 bit 数，同一个 block 中每个值所需 bit 数相同
meta.writeLong(min); // 最小值
meta.writeLong(gcd); // 最大公约数
long startOffset = data.getFilePointer();
meta.writeLong(startOffset);
if (doBlocks) {
// 写多个 block
writeValuesMultipleBlocks(valuesProducer.getSortedNumeric(field), gcd);
} else if (numBitsPerValue != 0) {
// 写单个 block
writeValuesSingleBlock(valuesProducer.getSortedNumeric(field), numValues, numBitsPerValue, min, gcd, encode);
}
meta.writeLong(data.getFilePointer() - startOffset);
return new long[] {numDocsWithValue, numValues};
}在写单个或多个 block 的时候都会初始化一个 DirectWriter 来执行直接按 byte 写的逻辑，该函数的构造方法：
DirectWriter.java
DirectWriter(DataOutput output, long numValues, int bitsPerValue) {
this.output = output;
this.numValues = numValues;
this.bitsPerValue = bitsPerValue;
encoder = BulkOperation.of(PackedInts.Format.PACKED, bitsPerValue);
iterations = encoder.computeIterations((int) Math.min(numValues, Integer.MAX_VALUE), PackedInts.DEFAULT_BUFFER_SIZE);// 计算在不超过 1k 内存的情况下需要多少轮迭代
nextBlocks = new byte[iterations * encoder.byteBlockCount()]; // byteBlockCount: 多少个 byte 存 bitsPerValue 对象，例如 bitsPerValue = 24，则 byteBlockCount = 24/8=3
nextValues = new long[iterations * encoder.byteValueCount()]; // byteValueCount: byteBlockCount 个 byte 能存多少个 value
/**
举例如下：
* * - 16 bits per value -> b=2, v=1 2*8 = 16/16 = 1
* - 24 bits per value -> b=3, v=1 3*8 = 24/24 = 1
* - 50 bits per value -> b=25, v=4 25*8 = 200/50 = 4
* - 63 bits per value -> b=63, v=8 63*8 = 504/63 = 8
*/
}写单个 block 的逻辑，在下面的 writer.add 函数中添加值到内部的 nextValues 数组中（数组长度就是上面的 iterations * byteValueCount），满了就逐个 byte 刷一次盘。
Lucene70DocValuesConsumer.java
private void writeValuesSingleBlock(SortedNumericDocValues values, long numValues, int numBitsPerValue,
long min, long gcd, Map<Long, Integer> encode) throws IOException {
DirectWriter writer = DirectWriter.getInstance(data, numValues, numBitsPerValue);
for (int doc = values.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = values.nextDoc()) {
for (int i = 0, count = values.docValueCount(); i < count; ++i) {
long v = values.nextValue();
if (encode == null) {
// 值减掉最小值再除以最大公约数
writer.add((v - min) / gcd);
} else {
// 很多 unique value，保存 meta 中存的 value 的位置
writer.add(encode.get(v));
}
}
}
writer.finish();
}
写多个 block 的场景，只是按 block 分开保存相应的 bitPerValue，以及meta 中多一些标记位。目的是为了降低存储空间。特别是值的大小差异很大的时候，拆分成多个 block 每个 block 按照自己的 bitPerValue 要比直接按整个 segment 所有 value 算 bitPerValue 节省空间。可以参考前面文件结构中 multiple block 写的场景结构，以及 Lucene70DocValuesConsumer 类的 Lucene70DocValuesConsumer 函数。
前面是 SortedNumericDocValuesWriter 的刷新逻辑，接下来我们看一下 SortedSetDocValuesWriter 的刷新逻辑。它主要处理字符数组类型的字段。SortedSet 字段默认会将 value 按 byte 排序，并生成新的 docId 映射，见下面 flush 函数中的 ordMap：
SortedSetDocValuesWriter.java
@Override
public void flush(SegmentWriteState state, Sorter.DocMap sortMap, DocValuesConsumer dvConsumer) throws IOException {
......
ords = pending.build(); // 每个值在 hash 中对应的位置，和 docId 顺序一致
ordCounts = pendingCounts.build(); // 数组的场景，记录该文档该字段中的值数量
sortedValues = hash.sort(); // 对值进行排序，返回值对应的新的位置列表，此 hash 中既保存的了原始的 bytes，也保存的位置
ordMap = new int[valueCount];
for(int ord=0;ord<valueCount;ord++) { // 这里对排好序的位置做一个映射，映射之后的 ordMap 顺序和 docId 顺序一致
ordMap[sortedValues[ord]] = ord;
}
......SortedSet 字段写 dvd、dvm 的逻辑主要在 Lucene70DocValuesConsumer.doAddSortedField 函数中。主要分为三层，第一层是每个 value 的 hash 位置，第二层是每个 value 的原始值，第三层是原始值的索引项。每层依次保存，并有对应的偏移量保存在元数据中。
第一层：
Lucene70DocValuesConsumer.java
private void doAddSortedField(FieldInfo field, DocValuesProducer valuesProducer) throws IOException {
......
values = valuesProducer.getSorted(field);
for (int doc = values.nextDoc(); doc != DocIdSetIterator.NO_MORE_DOCS; doc = values.nextDoc()) {
writer.add(values.ordValue()); // 第一层，这里写入的是每个 value 对应 hash 中的位置信息
}
writer.finish();
meta.writeLong(data.getFilePointer() - start); // 元数据保存偏移量
......
// 第二层，添加每个 value 的 term，保存原始值及索引
addTermsDict(DocValues.singleton(valuesProducer.getSorted(field)));
}第二层逻辑：
Lucene70DocValuesConsumer.java
/**
* SortedSet 对象，这里保存 value 的 terms dict，采用前缀压缩方法
* @param values
* @throws IOException
*/
private void addTermsDict(SortedSetDocValues values) throws IOException {
final long size = values.getValueCount();
meta.writeVLong(size);
meta.writeInt(Lucene70DocValuesFormat.TERMS_DICT_BLOCK_SHIFT); // 划分 block，一个 block 最大16个对象
RAMOutputStream addressBuffer = new RAMOutputStream();
meta.writeInt(DIRECT_MONOTONIC_BLOCK_SHIFT);
long numBlocks = (size + Lucene70DocValuesFormat.TERMS_DICT_BLOCK_MASK) >>> Lucene70DocValuesFormat.TERMS_DICT_BLOCK_SHIFT;// values 切成多少个 block
DirectMonotonicWriter writer = DirectMonotonicWriter.getInstance(meta, addressBuffer, numBlocks, DIRECT_MONOTONIC_BLOCK_SHIFT);
BytesRefBuilder previous = new BytesRefBuilder();
long ord = 0;
long start = data.getFilePointer();
int maxLength = 0;
TermsEnum iterator = values.termsEnum();
for (BytesRef term = iterator.next(); term != null; term = iterator.next()) {
if ((ord & Lucene70DocValuesFormat.TERMS_DICT_BLOCK_MASK) == 0) {// block 满了记录长度，当前 term 直接写入
writer.add(data.getFilePointer() - start); // 这里记录每个 block 的长度，会作数值压缩保存并记录 meta， data 先存 addressBuffer ，稍后写入 data 文件
data.writeVInt(term.length);
data.writeBytes(term.bytes, term.offset, term.length);
} else {
final int prefixLength = StringHelper.bytesDifference(previous.get(), term);// 和前值比较，计算出相同前缀长度
final int suffixLength = term.length - prefixLength; // 后缀长度
assert suffixLength > 0; // terms are unique
// 用一个 byte 的高4位和低4位分别保存前后缀长度，如果前缀超过15，或者后缀超过16，单独记录超过数量
data.writeByte((byte) (Math.min(prefixLength, 15) | (Math.min(15, suffixLength - 1) << 4)));
if (prefixLength >= 15) {
data.writeVInt(prefixLength - 15);
}
if (suffixLength >= 16) {
data.writeVInt(suffixLength - 16);
}
data.writeBytes(term.bytes, term.offset + prefixLength, term.length - prefixLength); // 写后缀内容
}
maxLength = Math.max(maxLength, term.length);
previous.copyBytes(term); // 保存当前值便于和下一个值比较
++ord;
}
writer.finish();
meta.writeInt(maxLength); // value 的最大长度
meta.writeLong(start); // 起始位置
meta.writeLong(data.getFilePointer() - start); // 结束位置
start = data.getFilePointer();
addressBuffer.writeTo(data); // 将每个 block 的长度信息写入 data 文件
meta.writeLong(start); // 写入长度信息的起始位置
meta.writeLong(data.getFilePointer() - start); // 写入长度信息的结束位置
// 第三层，记录 term 字典的索引，values 是按照值 hash 排过序的，这里每 1024 条抽取一个作为索引，加速查询
writeTermsIndex(values);
}第三层逻辑：
Lucene70DocValuesConsumer.java
private void writeTermsIndex(SortedSetDocValues values) throws IOException {
final long size = values.getValueCount();
meta.writeInt(Lucene70DocValuesFormat.TERMS_DICT_REVERSE_INDEX_SHIFT); // 索引抽取粒度，1024
long start = data.getFilePointer();
long numBlocks = 1L + ((size + Lucene70DocValuesFormat.TERMS_DICT_REVERSE_INDEX_MASK) >>> Lucene70DocValuesFormat.TERMS_DICT_REVERSE_INDEX_SHIFT);
RAMOutputStream addressBuffer = new RAMOutputStream();
DirectMonotonicWriter writer = DirectMonotonicWriter.getInstance(meta, addressBuffer, numBlocks, DIRECT_MONOTONIC_BLOCK_SHIFT);
TermsEnum iterator = values.termsEnum();
BytesRefBuilder previous = new BytesRefBuilder();
long offset = 0;
long ord = 0;
for (BytesRef term = iterator.next(); term != null; term = iterator.next()) {
if ((ord & Lucene70DocValuesFormat.TERMS_DICT_REVERSE_INDEX_MASK) == 0) {
writer.add(offset);
final int sortKeyLength;
if (ord == 0) {
// no previous term: no bytes to write
sortKeyLength = 0;
} else {
sortKeyLength = StringHelper.sortKeyLength(previous.get(), term);
}
offset += sortKeyLength;
data.writeBytes(term.bytes, term.offset, sortKeyLength); // 索引项也采用前缀压缩
} else if ((ord & Lucene70DocValuesFormat.TERMS_DICT_REVERSE_INDEX_MASK) == Lucene70DocValuesFormat.TERMS_DICT_REVERSE_INDEX_MASK) {
previous.copyBytes(term); // 每到达 1024 的位置抽取值
}
++ord;
}
writer.add(offset);
writer.finish();
meta.writeLong(start); // 保存索引项的起始位置
meta.writeLong(data.getFilePointer() - start); // 保存索引项总的长度
start = data.getFilePointer();
addressBuffer.writeTo(data); // 保存每个索引项的长度信息
meta.writeLong(start); // 索引项长度起始位置
meta.writeLong(data.getFilePointer() - start); // 索引项长度信息的总大小
}以上就是 SortedSet 类型的刷新落盘逻辑。至此，整个写入、刷新流程就分析到这里，接下来继续看合并流程。
合并流程
合并流程逻辑主要是读取待合并的每个 segment 的 doc value，然后在做一次写入流程。调用时序如下：
周期性的合并或者 indexing 过程中的合并，最终的入口在 SegmentMerger.merge()，里面包含各个数据结构的合并逻辑，segmentWriteState 包含了待 merge 的所有 segment 信息。简化之后的代码：
SegmentMerger.java
MergeState merge() throws IOException {
mergeTerms(segmentWriteState);
if (mergeState.mergeFieldInfos.hasDocValues()) {
mergeDocValues(segmentWriteState); // doc value 的合并
}
if (mergeState.mergeFieldInfos.hasPointValues()) {
mergePoints(segmentWriteState);
}
if (mergeState.mergeFieldInfos.hasNorms()) {
mergeNorms(segmentWriteState);
}
if (mergeState.mergeFieldInfos.hasVectors()) {
numMerged = mergeVectors();
}
// write the merged infos
codec.fieldInfosFormat().write(directory, mergeState.segmentInfo, "", mergeState.mergeFieldInfos, context);
return mergeState;
}mergeDocValues 会调用 DocValuesConsumer.merge 函数，遍历每个 field 在各 segement 里面的 doc values，逐个读取在内存中合并，然后写入新的 segment。
DocValuesConsumer.java
public void merge(MergeState mergeState) throws IOException {
for (FieldInfo mergeFieldInfo : mergeState.mergeFieldInfos) {
DocValuesType type = mergeFieldInfo.getDocValuesType();
if (type != DocValuesType.NONE) {
if (type == DocValuesType.NUMERIC) {
mergeNumericField(mergeFieldInfo, mergeState);
} else if (type == DocValuesType.BINARY) {
mergeBinaryField(mergeFieldInfo, mergeState);
} else if (type == DocValuesType.SORTED) {
mergeSortedField(mergeFieldInfo, mergeState);
} else if (type == DocValuesType.SORTED_SET) {
mergeSortedSetField(mergeFieldInfo, mergeState);
} else if (type == DocValuesType.SORTED_NUMERIC) {
mergeSortedNumericField(mergeFieldInfo, mergeState);
} else {
throw new AssertionError("type=" + type);
}
}
}
}例如，合并 numeric field：
DocValuesConsumer.java
public void mergeNumericField(final FieldInfo mergeFieldInfo, final MergeState mergeState) throws IOException {
addNumericField(mergeFieldInfo, // 调 Lucene70DocValuesConsumer 的写入逻辑
new EmptyDocValuesProducer() {
@Override
public NumericDocValues getNumeric(FieldInfo fieldInfo) throws IOException {
for (int i=0;i<mergeState.docValuesProducers.length;i++) { // 遍历该 field 在每个 segment 里面的 doc value
NumericDocValues values = null;
DocValuesProducer docValuesProducer = mergeState.docValuesProducers[i];
if (docValuesProducer != null) {
FieldInfo readerFieldInfo = mergeState.fieldInfos[i].fieldInfo(mergeFieldInfo.name);
if (readerFieldInfo != null && readerFieldInfo.getDocValuesType() == DocValuesType.NUMERIC) {
values = docValuesProducer.getNumeric(readerFieldInfo);
}
}
if (values != null) {
cost += values.cost();
subs.add(new NumericDocValuesSub(mergeState.docMaps[i], values)); // 合并稍后一起读取
}
}
......
}读取流程
在 ES 节点启动之后，会读取 segment meta data，之后在需要查询某个字段的 doc value 的时候，会先将对应的内容映射到内存，然后顺序获取对应的值。如果是字符或字符数组类型，则还会调用获取 hash 值位置以及对应 term 的函数得到原始数据。在排序、聚合、范围查询等场景可能会使用到 doc value，这取决于对应查询条件的 cost 权重。
读取逻辑的代码几乎都在 Lucene70DocValuesProducer 类中，这里就不展开描述了，大家可以对照上述调用时序看一下代码。
至此，doc value 的写入、合并、读取流程及其文件数据结构就分析完了，本文只分析了主要的正常流程，暂未考虑其它异常分支流程。欢迎各位提出意见，一起交流学习！
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
腾讯云开发者
