Paimon的部分更新以及DeleteVector实现
背景
本文基于 Paimon 0.9
出于对与Paimon内部的DeleteVctor的实现以及部分更新的实现进行的源码阅读。
关于 DeleteVector的介绍可以看这里
说明
对于Paimon来说无论是Spark中使用还是Flink使用，后面的逻辑都是一样的，所以我们以Spark为例来说。所以我们会参考类 org.apache.paimon.spark.SparkSource,
对于Flink可以参考org.apache.paimon.flink.FlinkTableFactory
如没特别说明，这里都是以主键表来进行说明。
paimon的部分字段更新
这里主要的场景更多的是多流或者多批写同一个表字段的场景，且每个流或批只更新某几个字段（同样的主键），具体的配置或说明参考Partial Update
这里涉及到的方法为 SparkTableWrite.write,最终会到MergeTreeWriter.write:
@Overridepublic void write(KeyValue kv) throws Exception {long sequenceNumber = newSequenceNumber();boolean success = writeBuffer.put(sequenceNumber, kv.valueKind(), kv.key(), kv.value());if (!success) {flushWriteBuffer(false, false);success = writeBuffer.put(sequenceNumber, kv.valueKind(), kv.key(), kv.value());if (!success) {throw new RuntimeException("Mem table is too small to hold a single element.");}}}
writeBuffer.put主要是往buffer中写数据
这里的writeBuffer是SortBufferWriteBuffer类实例。
这里会 主键+sequenceNumber+valueKind + value 的形式写入数据flushWriteBuffer这里就会涉及到数据落盘以及部分更新的逻辑：writeBuffer.forEach(keyComparator,mergeFunction,changelogWriter == null ? null : changelogWriter::write,dataWriter::write);- mergeFunction 这里的函数就是会在
MergeTreeWriter初始化，也就是会初始化为PartialUpdateMergeFunction。 - 对于
forEach的实现会构建一个MergeIterator,在这里面会调用PartialUpdateMergeFunction.add方法
这里就会涉及到部分更新的逻辑，主要就是：把按照主键+sequenceNumber排序好的数据传给PartialUpdateMergeFunction，
这样PartialUpdateMergeFunction只需要判断前后两个的数据的主键是否一致来进行更新。
具体的更新逻辑见: Partial Update
这里的new MergeIterator(awConsumer, buffer.sortedIterator(), keyComparator, mergeFunction);buffer.sortedIterator主要看SortBufferWriteBuffer构造方法（也就是为什么会按照主键+sequenceNumber排序）：
其中public SortBufferWriteBuffer(RowType keyType,RowType valueType,@Nullable FieldsComparator userDefinedSeqComparator,MemorySegmentPool memoryPool,boolean spillable,MemorySize maxDiskSize,int sortMaxFan,CompressOptions compression,IOManager ioManager) {...// key fieldsIntStream sortFields = IntStream.range(0, keyType.getFieldCount());// user define sequence fieldsif (userDefinedSeqComparator != null) {IntStream udsFields =IntStream.of(userDefinedSeqComparator.compareFields()).map(operand -> operand + keyType.getFieldCount() + 2);sortFields = IntStream.concat(sortFields, udsFields);}// sequence fieldsortFields = IntStream.concat(sortFields, IntStream.of(keyType.getFieldCount()));int[] sortFieldArray = sortFields.toArray();// row typeList<DataType> fieldTypes = new ArrayList<>(keyType.getFieldTypes());fieldTypes.add(new BigIntType(false));fieldTypes.add(new TinyIntType(false));fieldTypes.addAll(valueType.getFieldTypes());NormalizedKeyComputer normalizedKeyComputer =CodeGenUtils.newNormalizedKeyComputer(fieldTypes, sortFieldArray);RecordComparator keyComparator =CodeGenUtils.newRecordComparator(fieldTypes, sortFieldArray);...InternalRowSerializer serializer =InternalSerializers.create(KeyValue.schema(keyType, valueType));BinaryInMemorySortBuffer inMemorySortBuffer =BinaryInMemorySortBuffer.createBuffer(normalizedKeyComputer, serializer, keyComparator, memoryPool);IntStream.concat(sortFields, IntStream.of(keyType.getFieldCount()))就会会把sequenceNumber这个字段带入到排序中去，
也就是在buffer.sortedIterato方法中调用。
如果有定义sequence.field，那这里面的字段也会参与排序，见：udsFields字段
- mergeFunction 这里的函数就是会在
DeleteVector的实现
关于deleteVector的实现，可以参考Introduce deletion vectors for primary key table
大概的思想是： 基于Compaction + lookup的机制产生 DeleteVector:
- 当一个记录不属于 level0层的话，就不会产生DelectVector
- 当一个记录只属于需要进行compaction的level的话，就不会产生DeleteVector
- 当一个记录只属于 level0层的话，就要去查询不包含 Compaction的层的文件数据，从而产生DeleteVector
注意： deleteVector只支持主键表， 是属于bucket级别的，一个bucket一个DeleteVector。
DeleteVector的写
按照以上的说法，只有在Compaction的时候，才会产生DeleteVector，所以 我们直接到达 MergeTreeWriter.flushWriteBuffer,这里涉及到DeleteVector的数据流如下：
compactManager.triggerCompaction(forcedFullCompaction)||\/
submitCompaction||\/
MergeTreeCompactTask.doCompact||\/rewrite ||\/
rewriteImpl ||\/
LookupMergeTreeCompactRewriter.rewrite ||\/
rewriteOrProduceChangelog||\/
createMergeWrapper||\/
iterator.next()||\/
RecordReaderIterator.next()||\/
advanceIfNeeded||\/
currentIterator.next() ||\/
SortMergeIterator.next()||\/
LookupChangelogMergeFunctionWrapper.add(winner)||\/
LookupChangelogMergeFunctionWrapper.getResult()-
这里
MergeTreeCompactTask.doCompact写完之后，会有result.setDeletionFile(compactDfSupplier.get())
compactDfSupplier这里的源自submitCompaction方法中的compactDfSupplier构造:if (dvMaintainer != null) {compactDfSupplier =lazyGenDeletionFile? () -> CompactDeletionFile.lazyGeneration(dvMaintainer): () -> CompactDeletionFile.generateFiles(dvMaintainer);}而这里的deleteVector的产生来自
LookupChangelogMergeFunctionWrapper.getResult()，见以下说明 -
这里的
LookupMergeTreeCompactRewriter.rewrite的LookupMergeTreeCompactRewriter实例是在创建MergeTreeWriter的CompactManager compactManager =createCompactManager(partition, bucket, compactStrategy, compactExecutor, levels, dvMaintainer)这里会调用
createRewriter方法创建LookupMergeTreeCompactRewriter实例，
其中会根据lookupStrategy来创建该实例:public LookupStrategy lookupStrategy() {return LookupStrategy.from(mergeEngine().equals(MergeEngine.FIRST_ROW),changelogProducer().equals(ChangelogProducer.LOOKUP),deletionVectorsEnabled(),options.get(FORCE_LOOKUP)); -
这里
currentIterator.next()是 通过调用currentIterator = SortMergeReaderWithLoserTree.readBatch获取的,而SortMergeReaderWithLoserTree是通过readerForMergeTree方法获取的 -
这里
LookupChangelogMergeFunctionWrapper.getResult()才是重点@Overridepublic ChangelogResult getResult() {// 1. Compute the latest high level record and containLevel0 of candidatesLinkedList<KeyValue> candidates = mergeFunction.candidates();Iterator<KeyValue> descending = candidates.descendingIterator();KeyValue highLevel = null;boolean containLevel0 = false;while (descending.hasNext()) {KeyValue kv = descending.next();if (kv.level() > 0) {descending.remove();if (highLevel == null) {highLevel = kv;}} else {containLevel0 = true;}}// 2. Lookup if latest high level record is absentif (highLevel == null) {InternalRow lookupKey = candidates.get(0).key();T lookupResult = lookup.apply(lookupKey);if (lookupResult != null) {if (lookupStrategy.deletionVector) {PositionedKeyValue positionedKeyValue = (PositionedKeyValue) lookupResult;highLevel = positionedKeyValue.keyValue();deletionVectorsMaintainer.notifyNewDeletion(positionedKeyValue.fileName(), positionedKeyValue.rowPosition());} else {highLevel = (KeyValue) lookupResult;}}}// 3. Calculate resultKeyValue result = calculateResult(candidates, highLevel);// 4. Set changelog when there's level-0 recordsreusedResult.reset();if (containLevel0 && lookupStrategy.produceChangelog) {setChangelog(highLevel, result);}return reusedResult.setResult(result);} -
这里主要说明
lookup.apply的方法，其中 lookup的 构造是在create中LookupChangelogMergeFunctionWrapper构造中:@Overridepublic MergeFunctionWrapper<ChangelogResult> create(MergeFunctionFactory<KeyValue> mfFactory,int outputLevel,LookupLevels<T> lookupLevels,@Nullable DeletionVectorsMaintainer deletionVectorsMaintainer) {return new LookupChangelogMergeFunctionWrapper<>(mfFactory,key -> {try {return lookupLevels.lookup(key, outputLevel + 1);} catch (IOException e) {throw new UncheckedIOException(e);}},valueEqualiser,changelogRowDeduplicate,lookupStrategy,deletionVectorsMaintainer,userDefinedSeqComparator);}}这里的
lookupLevels.lookup会最终调用createLookupFile方法构造LookupFile实例,
其中会调用valueProcessor.persistToDisk(kv, batch.returnedPosition()方法，持久化 行号到对应的文件，
这样就能获取到对应的行号。 -
获取到对应的结果 lookupResult 后
调用deletionVectorsMaintainer.notifyNewDeletion(positionedKeyValue.fileName(), positionedKeyValue.rowPosition()方法去构造
DeletionVector.
上面提到的result.setDeletionFile(compactDfSupplier.get())会调用CompactDeletionFile.generateFiles(dvMaintainer)方法
从而调用maintainer.writeDeletionVectorsIndex方法，从而写如到DeleteVector文件中。
DeleteVector的读
DeleteVector的读取主要在以下方法中构造：PrimaryKeyFileStoreTable.newRead:
最终会调用RawFileSplitRead.createReader从而调用 ApplyDeletionVectorReader(fileRecordReader, deletionVector)方法构造ApplyDeletionVectorReader实例：
public RecordIterator<InternalRow> readBatch() throws IOException {RecordIterator<InternalRow> batch = reader.readBatch();if (batch == null) {return null;}checkArgument(batch instanceof FileRecordIterator,"There is a bug, RecordIterator in ApplyDeletionVectorReader must be RecordWithPositionIterator");return new ApplyDeletionFileRecordIterator((FileRecordIterator<InternalRow>) batch, deletionVector);}
该处的readBatch方法会构造一个ApplyDeletionFileRecordIterator迭代器,可见在next()方法会对每一个记录调用deletionVector.isDeleted是否删除的判断.
@Overridepublic InternalRow next() throws IOException {while (true) {InternalRow next = iterator.next();if (next == null) {return null;}if (!deletionVector.isDeleted(returnedPosition())) {return next;}}}
FAQ
写入文件的时候，怎么记录行号和主键的关系？
这里不会写入的时候记录行号，会在调用createLookupFile 在构建 LookupFile这个文件的时候(初始化)，从parquet文件读取过来的时候，就会获取行号。