Lookup Store 主要用于 Paimon 中的 Lookup Compaction 以及 Lookup join 的场景. 会将远程的列存文件在本地转化为 KV 查找的格式.
Hash
https://github.com/linkedin/PalDB
Sort
https://github.com/dain/leveldb
https://github.com/apache/paimon/pull/3770
整体文件结构:
相比于 Hash file 的优势
- 一次写入, 避免了文件merge
- 顺序写入, 保持原先的 key 的顺序, 后续如果按照 key 的顺序查找, 可提升缓存效率
SortLookupStoreWriter
SortLookupStoreWriter#put
put
@Override
public void put(byte[] key, byte[] value) throws IOException {dataBlockWriter.add(key, value);if (bloomFilter != null) {bloomFilter.addHash(MurmurHashUtils.hashBytes(key));}lastKey = key;// 当BlockWriter写入达到一定阈值, 默认是 cache-page-size=64kb.if (dataBlockWriter.memory() > blockSize) {flush();}recordCount++;
}
flush
private void flush() throws IOException { if (dataBlockWriter.size() == 0) { return; } // 将data block写入数据文件, 并记录对应的position和长度BlockHandle blockHandle = writeBlock(dataBlockWriter); MemorySlice handleEncoding = writeBlockHandle(blockHandle);// 将BlockHandle 写入index writer, 这也通过是一个BlockWriter写的indexBlockWriter.add(lastKey, handleEncoding.copyBytes());
}
writeBlock
private BlockHandle writeBlock(BlockWriter blockWriter) throws IOException {// close the block// 获取block的完整数组, 此时blockWriter中的数组并不会被释放, 而是会继续复用MemorySlice block = blockWriter.finish();totalUncompressedSize += block.length();// attempt to compress the blockBlockCompressionType blockCompressionType = BlockCompressionType.NONE;if (blockCompressor != null) {int maxCompressedSize = blockCompressor.getMaxCompressedSize(block.length());byte[] compressed = allocateReuseBytes(maxCompressedSize + 5);int offset = encodeInt(compressed, 0, block.length());int compressedSize =offset+ blockCompressor.compress(block.getHeapMemory(),block.offset(),block.length(),compressed,offset);// Don't use the compressed data if compressed less than 12.5%,if (compressedSize < block.length() - (block.length() / 8)) {block = new MemorySlice(MemorySegment.wrap(compressed), 0, compressedSize);blockCompressionType = this.compressionType;}}totalCompressedSize += block.length();// create block trailer// 每一块block会有一个trailer, 记录压缩类型和crc32校验码BlockTrailer blockTrailer =new BlockTrailer(blockCompressionType, crc32c(block, blockCompressionType));MemorySlice trailer = BlockTrailer.writeBlockTrailer(blockTrailer);// create a handle to this block// BlockHandle 记录了每个block的其实position和长度BlockHandle blockHandle = new BlockHandle(position, block.length());// write data// 将数据追加写入磁盘文件writeSlice(block);// write trailer: 5 bytes// 写出trailerwriteSlice(trailer);// clean up stateblockWriter.reset();return blockHandle;
}
close
public LookupStoreFactory.Context close() throws IOException {// flush current data blockflush();LOG.info("Number of record: {}", recordCount);// write bloom filter@Nullable BloomFilterHandle bloomFilterHandle = null;if (bloomFilter != null) {MemorySegment buffer = bloomFilter.getBuffer();bloomFilterHandle =new BloomFilterHandle(position, buffer.size(), bloomFilter.expectedEntries());writeSlice(MemorySlice.wrap(buffer));LOG.info("Bloom filter size: {} bytes", bloomFilter.getBuffer().size());}// write index block// 将index数据写出至文件BlockHandle indexBlockHandle = writeBlock(indexBlockWriter);// write footer// Footer 记录bloomfiler + indexFooter footer = new Footer(bloomFilterHandle, indexBlockHandle);MemorySlice footerEncoding = Footer.writeFooter(footer);writeSlice(footerEncoding);// 最后关闭文件// close filefileOutputStream.close();LOG.info("totalUncompressedSize: {}", MemorySize.ofBytes(totalUncompressedSize));LOG.info("totalCompressedSize: {}", MemorySize.ofBytes(totalCompressedSize));return new SortContext(position);
}
BlockWriter
add
public void add(byte[] key, byte[] value) {int startPosition = block.size();// 写入key长度block.writeVarLenInt(key.length);// 写入keyblock.writeBytes(key);// 写入value长度block.writeVarLenInt(value.length);// 写入valueblock.writeBytes(value);int endPosition = block.size();// 使用一个int数组记录每个KV pair的起始位置作为索引positions.add(startPosition);// 是否对齐. 是否对齐取决于每个KV对的长度是否一样if (aligned) {int currentSize = endPosition - startPosition;if (alignedSize == 0) {alignedSize = currentSize;} else {aligned = alignedSize == currentSize;}}
}
- 这里的 block 对应于一块可扩容的 MemorySegment, 也就是
byte[], 当写入长度超过当前数组的长度时, 就会扩容
finish
public MemorySlice finish() throws IOException {if (positions.isEmpty()) {throw new IllegalStateException();}// 当通过BlockWriter写出的数据长度都是对齐的时, 就不需要记录各个Position的index了, 只需要记录一个对齐长度, 读取时自己可以计算.if (aligned) {block.writeInt(alignedSize);} else {for (int i = 0; i < positions.size(); i++) {block.writeInt(positions.get(i));}block.writeInt(positions.size());}block.writeByte(aligned ? ALIGNED.toByte() : UNALIGNED.toByte());return block.toSlice();
}
小结
整个文件的写出过程非常简单, 就是按 block 写出, 并且记录每个 block 的位置, 作为 index.
SortLookupStoreReader
读取的过程, 主要就是为了查找 key 是否存在, 以及对应的 value 或者对应的行号.
public byte[] lookup(byte[] key) throws IOException {// 先通过bloomfilter提前进行判断if (bloomFilter != null && !bloomFilter.testHash(MurmurHashUtils.hashBytes(key))) {return null;}MemorySlice keySlice = MemorySlice.wrap(key);// seek the index to the block containing the keyindexBlockIterator.seekTo(keySlice);// if indexIterator does not have a next, it means the key does not exist in this iteratorif (indexBlockIterator.hasNext()) {// seek the current iterator to the key// 根据从index block中读取到的key value的位置(BlockHandle), 读取对应的value blockBlockIterator current = getNextBlock();// 在value的iterator中再次二分查找寻找对应block中是否存在match的key, 如果存在则返回对应的数据if (current.seekTo(keySlice)) {return current.next().getValue().copyBytes();}}return null;
}
- 查找一次 key 会经历两次二分查找(index + value).
BlockReader
// 从block创建一个iterator
public BlockIterator iterator() {BlockAlignedType alignedType =BlockAlignedType.fromByte(block.readByte(block.length() - 1));int intValue = block.readInt(block.length() - 5);if (alignedType == ALIGNED) {return new AlignedIterator(block.slice(0, block.length() - 5), intValue, comparator);} else {int indexLength = intValue * 4;int indexOffset = block.length() - 5 - indexLength;MemorySlice data = block.slice(0, indexOffset);MemorySlice index = block.slice(indexOffset, indexLength);return new UnalignedIterator(data, index, comparator);}
}
SliceCompartor
这里面传入了 keyComparator, 用于进行 key 的比较. 用于在 index 中进行二分查找. 这里的比较并不是直接基于原始的数据, 而是基于 MemorySlice 进行排序.
比较的过程会将 key 的各个字段从 MemorySegment 中读取反序列化出来, cast 成 Comparable 进行比较.
public SliceComparator(RowType rowType) {int bitSetInBytes = calculateBitSetInBytes(rowType.getFieldCount());this.reader1 = new RowReader(bitSetInBytes);this.reader2 = new RowReader(bitSetInBytes);this.fieldReaders = new FieldReader[rowType.getFieldCount()];for (int i = 0; i < rowType.getFieldCount(); i++) {fieldReaders[i] = createFieldReader(rowType.getTypeAt(i));}
}@Override
public int compare(MemorySlice slice1, MemorySlice slice2) {reader1.pointTo(slice1.segment(), slice1.offset());reader2.pointTo(slice2.segment(), slice2.offset());for (int i = 0; i < fieldReaders.length; i++) {boolean isNull1 = reader1.isNullAt(i);boolean isNull2 = reader2.isNullAt(i);if (!isNull1 || !isNull2) {if (isNull1) {return -1;} else if (isNull2) {return 1;} else {FieldReader fieldReader = fieldReaders[i];Object o1 = fieldReader.readField(reader1, i);Object o2 = fieldReader.readField(reader2, i);@SuppressWarnings({"unchecked", "rawtypes"})int comp = ((Comparable) o1).compareTo(o2);if (comp != 0) {return comp;}}}}return 0;
}
查找的实现就是二分查找的过程, 因为写入的 key 是有序写入的.
public boolean seekTo(MemorySlice targetKey) {int left = 0;int right = recordCount - 1;while (left <= right) {int mid = left + (right - left) / 2;// 对于aligned iterator, 就直接seek record * recordSize// 对于unaligned iterator, 就根据writer写入的索引表来跳转seekTo(mid);// 读取一条key value pairBlockEntry midEntry = readEntry();int compare = comparator.compare(midEntry.getKey(), targetKey);if (compare == 0) {polled = midEntry;return true;} else if (compare > 0) {polled = midEntry;right = mid - 1;} else {left = mid + 1;}}return false;
}
小结
查找过程
- 先过一遍 bloom filter
- index 索引查找对应 key 的 block handle
- 根据第二步的 handle, 读取对应的 block, 在 block 中查找对应的 key value.
