...
- 如果 compactionMergeWorking 变量为true,则说明上一次合并还在进行,跳过本次合并
- 如果 compactionMergeWorking 变量为false,说明本存储组没有合并任务在进行
- 设置 compactionMergeWorking 为true
- 拷贝当前文件列表镜像,见拷贝当前文件列表镜像,见 算法1
- 设置如果进行乱序合并,是否进行 FullMerge
- 提交合并任务
算法1
...
- 加读锁
- 按层拷贝当前时间分区除了最高层之外所有已经关闭的顺序文件
- 按层拷贝当前时间分区所有已经关闭的乱序文件
- 解读锁
FullMerge(完全合并)
出于IO和合并性能考虑,乱序合并不会重写整个顺序文件,获取一个整理完全的顺序文件,而是会将与乱序文件重叠的 chunk 合并重写后,追加到原顺序文件后面,不删除原chunk,称这个过程为 原地合并
...
而 完全合并 则是在乱序合并时重写整个顺序文件,IO和合并性能较低,但能获取一个整理完全的顺序文件
合并任务执行
- 选择文件并进行顺序空间的层级合并
- 如果有跨文件空间合并正在进行,等待该任务完成
- 对除了最高层外的文件
- 如果本层文件个数>seq
- _file_num_in_each_
- 对待合并文件列表中的每个文件,创建一个 TsFIleSequenceReader,并存在 ReaderCache 中,并获取该文件的设备列表,生成所有的 device 集合
- 遍历device集合
- 根据 ReaderCache 得到相应的TsFileSequenceReader
- 根据 device 读取各个文件对应的ChunkMetadataListIterator
- 循环算法2输出的待合并的 IMeasurementSchema 列表
- 对于待合并的 IMeasurementSchema
- 如果是乱序的文件合并,采取 存在对齐时间序列的反序列化 Page 合并算法
- 如果是顺序空间的文件合并
- 如果某个 Chunk(如果是对齐时间序列,判断 TimeChunk) 的数据点数小于 merge_page_point_number,采取 存在对齐时间序列的反序列化 Page 合并算法
- 如果page足够大,采取 存在对齐时间序列的追加 Page 合并算法
- 关闭 ReaderCache 中所有的reader
- 序列化新文件的 TsFileResource
- 关闭新文件writer
算法2
输入:多个文件的 ChunkMetadataListIterator,Iterator 每次输出的 List 内的 sensor 个数(包含对齐时间序列内的序列数,不包含 time)为 max_degree_of_index_node(由于对齐时间序列是个整体,因此可能会超过部分)
输出:每轮待合并的 sensor 列表
描述:每个迭代器取1个 List,找到每个 List 的最大字典序(对齐时间序列按 time 的名字 $#$id 来比较)的 sensor,组成集合 S,本次合并从头开始合并到 S 中最小字典序的 sensor 。并从 List 中清除已合并的 sensor,如果此 List 消耗完,获取下一个 Lst
优势:每次至少会消耗完一个 file 的一个 List
Iterator 每次输出的 List 的例子:文件内的序列为:(time, s4,s5,s6), (time, s9, s10), s1,s2,s3,s7,s8
如果 max_degree_of_index_node 为 2, 每次取出的为:(time, s4,s5,s6); (time, s9, s10); s1,s2; s3,s7; s8
如果 max_degree_of_index_node 为 3,每次取出的为:(time, s4,s5,s6); (time, s9, s10), s1; s2, s3,s7; s8
如果 max_degree_of_index_node 为 4,每次取出的为:(time, s4,s5,s6), (time, s9, s10); s1 s2, s3,s7; s8
存在对齐时间序列的反序列化 Page 合并算法
- 通过各自文件的IChunkReader有序把数据读出并整理出对应的time-value列表
- 如果是MeasurementSchema,使用ChunkReader
- 如果是VectorMeasurementSchema,使用VectorChunkReader
- 遍历上述的time-value列表,将数据写入新的IChunkWriter
- 如果是MeasurementSchema,使用ChunkWriterImpl
- 如果是VectorMeasurementSchema,使用VectorChunkWriterImpl
- 判断限流
- 将IChunkWriter写入新文件
存在对齐时间序列的追加 Page 合并算法
- 通过各自文件的reader有序把chunk读出来,并通过合并ByteBuffer和统计信息的方式不解析chunk数据点、而有序合并出对应的新Chunk和ChunkMetadata
- 如果是MeasurementSchema,仅需要前一个chunk和后一个chunk进行合并
- 如果是VectorMeasurementSchema,需要前一个VectorChunkMetadata对应的所有timeChunk和valueChunk与后一个所有的timeChunk和valueChunk合并
- 判断限流
- 将合并完的Chunk和ChunkMetadata写入新文件
例子:
VectorChunk1: timeChunk(page1,page2) s1Chunk(page3, page4) s2Chunk(page5)
VectorChunk2: timeChunk(page6) s1Chunk(page7) s2Chunk(page8)
合并后: timeChunk(page1, page2, page6) s1Chunk(page3, page4, page7), s2Chunk(page5, page8)
直接追加chunk合并
原流程存在的问题
- level,则进行将本层文件提交合并,详细合并流程见 合并机制执行合并流程文档-层级合并
- 选择文件并进行乱序空间的层级合并或跨文件空间合并
- 如果 unseq_level_num==1,提交当前时间分区下所有乱序文件和顺序文件进行合并
- 对除了最高层外的文件
- 如果本层文件个数>unseq_
...
- file_num_in_each_level
...
- 如果开启了跨文件空间的合并,且unseq_level_num-
...
- 2==当前层index
- 提交当前时间分区最高层的乱序文件和所有顺序文件进行跨文件空间合并,详细合并流程见 合并机制执行合并流程文档-跨文件空间合并
- 否则进行将本层文件提交层级合并,详细合并流程见 合并机制执行合并流程文档-层级合并
- 2==当前层index
1、在配置时需要知道用户原先写出的一个chunk的大小,并根据实际经验来配置。这样的配置方式对用户来说难以自己上手配置,对于每一个用户来说,也都需要我们去观察文件并辅助配置,非常不方便
2、一旦用户使用了默认参数没有修改或者配置错误,而用户本身场景又不需要合并太多次文件,那么就会占用很多无效的磁盘IO,甚至降低查询效率
3、在一些场景中,不同时间序列的写入速度是不一样的,那么按照这种合并方式,最后会合并出很多大小不一的chunk,写入速度慢的时间序列对应的chunk小,写入速度快的时间序列对应的chunk大,造成chunk大小不均匀,难以控制
例子:
假设用户需要的目标chunk是4倍大小的原始chunk,有如下文件
0层:file1(s1,s2,s2)+file2(s2,s2,s2) 即file1有1个s1的chunk,2个s2的chunk,file2有3个s2的chunk
合并至1层:file3(s1(1),s2(5)) 合并出来的文件有1个s1的chunk,1个5倍大小的s2的chunk,假设他与另一个file4(s1(1),s2(5))进行合并
合并至2层:file5(s1(2),s2(9)) 合并出来的文件有1个2倍大小的s1的chunk,1个9倍大小的s2的chunk
可以看到s2的chunk越合并越大,已经远远超过用户需要的chunk大小,s1对应的chunk却仍没有达到用户的要求
解决方案
使用配置merge_chunk_point_number_threshold在合并每一个chunk的时候控制,
如果在待合并列表中这个sensor对应的所有chunk都已经达到了这个阈值,则不再合并chunk,直接将读出来的chunk写入新文件
跨文件空间合并(将乱序文件合并至顺序空间)
不存在对齐时间序列的消除乱序文件合并流程
输入一组乱序文件U,以及一组顺序文件S,最大同时合并的时间序列数量为n,单个chunk的点数阈值tpt,是否进行Full Merge
...
7.1. 开启一个新的ChunkGroup;
7.2. 对于Tc中的每一条时间序列tsi,在S上查询它们的Chunk,并将这些Chunk按照在文件中的位置排序。
7.3. 取出位置最小的Chunk,记做c,如果已经没有Chunk,转7.4
7.3.1 找到c对应的MergeReader ri,如果ri当前的数据点的时间小于等于c的结束时间tend,将ri所有时间不大于tend的点和c合并,并将c写入到wi,转7.3.5;
7.3.2 如果c的点数小于tpt或者c的上一个Chunk已经写入到wi,但是wi还没有进行flush,将c写入到wi,转7.3.5;
7.3.3 如果进行Full Merge,将c写入到wi,转7.3.5;
7.3.4 记录这个没有被merge的Chunk,转7.3;
7.3.5 如果wi中已写入的点大于等于tpt,将wi flush到s'j,转7.3;
7.4 关闭这个ChunkGroup;
...
10.1 如果si中标记已被合并的chunk占总chunk的比例大于某值threshold:
10.1.0 在merge.log中记录“{si'} start {si'的当前长度}”
10.1.1 将si中未标记已被合并的chunk写入si';
10.1.2 为si'生成FileMetadata并写入到si'尾部;
10.1.3 等待对si的所有查询结束,并对si加锁;
10.1.4用si'替代si;
10.1.5 在merge.log中记录“{si} end”, 对si解锁;
10.2 否则:
10.2.1 等待对si的所有查询结束,并对si加锁;
10.2.2 将si的尾部的FileMetadata截去;
10.2.3 在merge.log中记录“{si} {si的当前长度}”
10.2.4 将si'中的数据写到si尾部;
10.2.5 为si生成FileMetadata并写入到si尾部,该FileMetadata中不包含被标记已经合并的chunk,但是记录有多少chunk被标为已经合并;
10.2.6 在merge.log中记录“{si} end”,删除si',对si解锁;
11. 在merge.log中记录“merge end”,删除U中的所有文件和merge.log
存在对齐时间序列的消除乱序文件合并流程
- 从MManager中的取出改storageGroup所有的device→IMeasurementSchema
- 遍历device列表 (去掉了sensor组的概念)
- 遍历顺序文件列表,对于每一个顺序文件
- 根据device读取各个文件对应的ChunkMetadataListIterator(ChunkMetadataListIterator每次按字典序吐出对应的IMeasurementSchema及其对应的ChunkMetadata列表,每批个数为1000个普通MeasurementSchema或超过1000个sensor的最小的完整的VectorMeasurementSchema列表)(这里需要改变底层结构提高性能)
- 循环遍历所有ChunkMetadataListIterator直到没有任何一个ChunkMetadataListIterator还有数据
- 对于每一个IMeasurementSchema建立IMeasurementSchema→chunkMetadataList的列表
- 读取每一个chunkMetadataList,建立List<List<Chunk>> chunks的结构
- 如果是ChunkMetadata,将当前chunk读出包裹一个List放入chunks
- 如果是VectorChunkMetadata,将当前chunk按timeChunk, valueChunk1,...valueChunkN 的顺序包裹一个List放入chunks
- 对于每一个IMeasurementSchema→chunkMetadataList列表进行按合并子任务并行配置进行分组
- 对于每个子任务的IMeasurementSchema→chunkMetadataList列表
- 按文件顺序遍历chunks列表的chunkList,对于chunkList的第一个chunk, 如果有与unseq文件overlapped数据,则插入IChunkWriter
- 如果IMeasurementSchema是MeasurementSchema,此时chunk列表的第一个chunk也是完整的chunk,直接写第一个chunk的数据写入ChunkWriterImpl
- 如果IMeasurementSchema是VectorMeasurementSchema,此时chunk列表的第一个chunk是timeChunk,需要将timeChunk, valueChunk1,...valueChunkN 所有本行数据写入 VectorChunkWriterImpl
如果遍历完该unseq文件还有剩余的数据,则直接append到IChunkWriter后面 - 如果IMeasurementSchema是MeasurementSchema,直接append第一个chunk的剩余数据 如果IMeasurementSchema是VectorMeasurementSchema,此时chunk列表的第一个chunk是timeChunk,需要将timeChunk, valueChunk1,...valueChunkN 所有剩余数据按行写入 VectorChunkWriterImpl
- 遍历顺序文件列表,对于每一个顺序文件