...

输入：selectedFiles（Map<TsFileResource, List<TsFileResource>>）

...

• for seqFile in selectedFiles.keySet()
  • 对 seqFile 建立一个 RestorableWriter
  • 获取与该 seqFile 进行合并的乱序文件列表 unseqFiles
  • 获取该 seqFile 的元数据列表 deviceChunkMetadataMap（Map<Device, Map<Sensor, List<ChunkMetadata>>>）
  • 获取 unseqFiles 中所有的时间序列元数据 Map<Device, List<Sensor>> deviceSensorMap
  • for device, sensors in deviceChunkMetadataMap
    • restorableWriter.startChunkGroup(device) 
    • for sensor in sensors
      • 为这个 sensor 在所有的 unseqFiles 上建立一个 unseqReader
      • 根据算法1将数据重写到 seqFile 中
    • restorableWriter.endChunkGroup(device)

...

• • 为unseqFiles增加 .mods 文件，并修改其 TsFileResource，抹去合并的数据

算法1

输入：待合并的 sensor

• if 这个 sensor 存在于 deviceChunkMetadataMap 中
  •  从 deviceChunkMetadataMap 中获取该 sensor 在 seqFile 中的 sensorChunkMetadataList
  • for chunkMetadata in sensorChunkMetadataList
    • chunk = readMemChunk(chunkMetadata)
    • unclosedChunkPoint = 0L
    • tsFileResource.updateStartTime(chunkMetadata.startTime)
    • tsFileResource.updateEndTime(chunkMetadata.endTime)
    • 判断当前 chunk 是否被修改 modified
    • if isOverlap(chunkMetadata)
      • 将其与乱序数据重叠的部分合并后写入 ChunkWriter（见 算法2）
      • unclosedChunkPoint+=写入的点数
    • else if isChunkTooSmall(chunkMetadata)
      • 将这个 Chunk 解压缩后写入 ChunkWriter
      • unclosedChunkPoint+=写入的点数
    • else
      • if  unclosedChunkPoint > 0 || modified
        
        • 将这个 Chunk 解压缩后写入 ChunkWriter
      • else 
        • 将这个 Chunk 不解压缩写入 writer
    • if unclosedChunkPoint > merge_chunk_point_num_threshold
      • 将 ChunkWriter 写入 writer
      • unclosedChunkPoint = 0
• 否则
  • 通过算法3将乱序数据写入 seqFile 中

算法2

输入：chunk, unseqReader, deviceEndTime, tsFileResource, modification

• 对 Chunk 构建一个 ChunkReader
• 使用 ChunkReader 获取 Chunk 中的每一个 Page
  • while pageData.hasNext() 
    • 获取 pageData 当前的时间戳 seqTime
    • overwriteSeqPoint = false
    • while 该 sensor 还有乱序数据并且乱序数据的下一个时间戳小于 seqTime
      • 将乱序数据写入 ChunkWrtier 中
      • tsFileResource.updateStartTime(乱序数据)
      • tsFileResource.updateEndTime(乱序数据)
      • 如果乱序数据的时间戳 == seqTime
        • overwriteSeqPoint = true
    • 根据 算法3 将该 unseqReader 小于等于 deviceEndTime 的剩余数据写入 ChunkWriter 中
    • if !overwriteSeqPoint && !pageData 当前时间戳在 modification 中被删除
      • 将 pageData 当前的数据点写入 ChunkWriter 中
      • tsFileResource.updateStartTime(乱序数据)
      • tsFileResource.updateEndTime(乱序数据)

算法3

输入：unseqReader、restorableWriter、seqFile

• 创建一个 ChunkWriter
• while unseqReader.currentTimestamp < seqFile.startTime
  • unseqReader.next()
• while unseqReader.currentTimestamp >= seqFile.startTime && unseqReader.currentTimestamp <= seqFile.endTime
  • 将 unseqReader 当前的时间点写入 ChunkWriter 中
  • 如果 ChunkWriter 中的数据点的个数 > CHUNK_MIN_POINT_NUM
    • flush ChunkWriter to restorableWriter
• 如果 ChunkWriter 中还有没刷盘的数据
  • flush chunkWriter to restorableWriter

...

碎文件清理

设计一个碎文件清理任务，将没有与任何顺序文件发生重叠的乱序文件填充到顺序空间中，详情见碎文件清理任务

...