Binary Row数据结构的实现

发表于 | 分类于 | | |
字数统计 2.7k | 阅读时长 13

[toc]

Binary Row是blink开源版本https://github.com/apache/flink/tree/blink中提到的一个runtime层面优化的特性，主要是应用于sql模块，简单来说，由于sql本身自带schema，在上下游数据传输的时候就可以利用这个schema信息来简化序列化和反序列化的过程，本文就来具体分析这个特性的实现。

主要实现代码在

1
2
3
flink-table org.apache.flink.table.typeutils
flink-table-common org.apache.flink.table.dataformat
flink-table-common org.apache.flink.table.typeutils

Binary Row

我们要理解sql层的数据传输是用的什么结构，只需要去观察runtime层实现的算子的传输数据类型即可，通过查看代码可以发现中间算子传输的均为BaseRow

1
public TwoInputSelection processElement1(StreamRecord<BaseRow> element) throws Exception {}

而之前版本的数据传输的是一个Row,内部是一个Object[]，在传输的过程中使用RowSerializer进行每一个字段的序列化和反序列化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public void serialize(Row record, DataOutputView target) throws IOException {
	int len = fieldSerializers.length;

	if (record.getArity() != len) {
		throw new RuntimeException("Row arity of from does not match serializers.");
	}

	// write a null mask
	writeNullMask(len, record, target);

	// serialize non-null fields
	for (int i = 0; i < len; i++) {
		Object o = record.getField(i);
		if (o != null) {
			fieldSerializers[i].serialize(o, target);
		}
	}
}

新的实现中以BaseRow代替了Row，baserow是一个基类，在不同的场景下有不同的子类去实现相应的功能。

GenericRow

能够方便的用以更新字段，其内部实现也是一个Object数组，

1
2
3
4
5
6
7
8
9
10
// kafka source deserialization schema 从source处就解析成一个BaseRow
public GenericRow deserialize(byte[] messageKey, byte[] message, String topic, int partition, long offset) throws IOException {
	GenericRow row = new GenericRow(5);
	row.update(0, messageKey);
	row.update(1, message);
	row.update(2, BinaryString.fromString(topic));
	row.update(3, partition);
	row.update(4, offset);
	return row;
}

JoinedRow

主要能够方便的将两个row进行拼接成一个baserow

1
2
// windowoperator中发送一个key和aggRes的组合的row到下游
reuseOutput.replace((BaseRow) getCurrentKey(), aggResult);

BinaryRow

BaseRow序列化是先转化成BinaryRow，然后再通过BinaryRowSerializer进行序列化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// BaseRowSerializer.java
public void serialize(BaseRow row, DataOutputView target) throws IOException {
	BinaryRow binaryRow;
	if (row.getClass() == BinaryRow.class) {
		binaryRow = (BinaryRow) row;
	} else {
		binaryRow = baseRowToBinary(row);
	}
	binarySerializer.serialize(binaryRow, target);
}

public BinaryRow baseRowToBinary(BaseRow baseRow) throws IOException {
	BinaryRow row = getProjection().apply(baseRow);
	row.setHeader(baseRow.getHeader());
	return row;
}

可以看到baseToRow的过程中首先是会通过codeGen生成映射函数，然后将baserow转成binaryrow，测试将以下的GenericRow转化成BinaryRow生成如下代码

1
2
3
4
5
6
7
GenericRow gR = new GenericRow(3);
gR.update(0, 1);
gR.update(1, 2L);
gR.update(2, "test");

BaseRowSerializer<GenericRow> serializer = new BaseRowSerializer<>(Types.INT, Types.LONG, Types.STRING);
BinaryRow row = serializer.baseRowToBinary(gR);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
public class BaseRowSerializerProjection$0 extends org.apache.flink.table.codegen.Projection<org.apache.flink.table.dataformat.BaseRow, org.apache.flink.table.dataformat.BinaryRow> {

        org.apache.flink.table.dataformat.BinaryString reuseBString$3 = new org.apache.flink.table.dataformat.BinaryString();
        final org.apache.flink.table.dataformat.BinaryRow out = new org.apache.flink.table.dataformat.BinaryRow(3);
        // 先构建一个BinaryRowWriter
        final org.apache.flink.table.dataformat.BinaryRowWriter outWriter = new org.apache.flink.table.dataformat.BinaryRowWriter(out);

        public BaseRowSerializerProjection$0() throws Exception {
          
        }

        @Override
        public org.apache.flink.table.dataformat.BinaryRow apply(org.apache.flink.table.dataformat.BaseRow in1) {
          int field$1;
          boolean isNull$1;
          long field$2;
          boolean isNull$2;
          org.apache.flink.table.dataformat.BinaryString field$4;
          boolean isNull$4;
          outWriter.reset();
          isNull$1 = in1.isNullAt(0);
          field$1 = -1;
          if (!isNull$1) {
            field$1 = in1.getInt(0);
          }
          if (isNull$1) {
            outWriter.setNullAt(0);
          } else {
            outWriter.writeInt(0, field$1);
          }
          isNull$2 = in1.isNullAt(1);
          field$2 = -1L;
          if (!isNull$2) {
            field$2 = in1.getLong(1);
          }
          if (isNull$2) {
            outWriter.setNullAt(1);
          } else {
            outWriter.writeLong(1, field$2);
          }
          isNull$4 = in1.isNullAt(2);
          field$4 = org.apache.flink.table.dataformat.BinaryString.EMPTY_UTF8;
          if (!isNull$4) {
            field$4 = in1.getBinaryString(2, reuseBString$3);
          }
          if (isNull$4) {
            outWriter.setNullAt(2);
          } else {
            outWriter.writeBinaryString(2, field$4);
          }
          outWriter.complete();
          return out;
        }
      }

通过codeGen的代码可以得出，对于普通的baserow，通过BinaryRowWriter，将baserow的每个字段写入到BinaryRow中，写入完成后，序列化的工作就都通过BinaryRowSerializer来完成。这样的好处有以下几个:

  • 如果某个中间算子只需要获取上游传输下来的某几个字段的值，那么只需要通过getXXX来直接获取，减少反序列化的量
  • 如果中间结果不发生改变，只需要将binaryRow直接拷贝出去，也减少了序列化的量

BinaryRow序列化的过程，可以看到就是直接的内存拷贝的过程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public void serialize(BinaryRow record, DataOutputView target) throws IOException {
	int sizeInBytes = record.getSizeInBytes();
	target.writeInt(sizeInBytes);
	int offset = record.getBaseOffset();
	for (MemorySegment segment : record.getAllSegments()) {
		int remain = segment.size() - offset;
		int copySize = remain > sizeInBytes ? sizeInBytes : remain;
		target.write(segment, offset, copySize);

		sizeInBytes -= copySize;
		offset = 0;
	}
	if (sizeInBytes != 0) {
		throw new RuntimeException("No copy finished, this should be a bug, " +
				"The remaining length is: " + sizeInBytes);
	}
}

BinaryString

在上面codegen的一段代码中，关于string的处理引入了一个概念BinaryString

1
2
3
4
field$4 = org.apache.flink.table.dataformat.BinaryString.EMPTY_UTF8;
if (!isNull$4) {
  field$4 = in1.getBinaryString(2, reuseBString$3);
}
1
2
3
4
5
6
7
8
9
// GenericRow的getBinaryString的实现
public BinaryString getBinaryString(int ordinal) {
	Object value = this.fields[ordinal];
	if (value instanceof BinaryString) {
		return (BinaryString) value;
	} else {
		return BinaryString.fromString((String) value);
	}
}

那么BinaryString是什么作用呢? 看注释

A utf8 string which is backed by {@link MemorySegment} instead of String. Its data may span multiple {@link MemorySegment}s.
一个直接存储在MemorySegment上的utf8的字符串，一个字符串数据可能会跨segment

1
2
3
4
5
6
private MemorySegment[] segments;
private int offset;
private int numBytes;

/** Cache the java string for the binary string to avoid redundant decode. */
private String javaString;

针对string类型，会在codegen阶段，将其转化成一个binaryString。从binarystring初始化的时候没有存储在MemorySegment之上，而是仅仅只是保存string的字符串信息，等到有对string的操作的时候,才会通过这个方法将其序列化，并包装成memorysegments

1
2
3
4
5
6
7
8
9
10
11
12
public void ensureEncoded() {
	if (!isEncoded()) {
		encodeToBytes();
	}
}

private void encodeToBytes() {
	if (javaString != null) {
		byte[] bytes = StringUtf8Utils.encodeUTF8(javaString);
		pointTo(bytes, 0, bytes.length, javaString);
	}
}

序列化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
public void serialize(BinaryString record, DataOutputView target) throws IOException {
	byte[] bytes = record.getBytes();
	target.writeInt(bytes.length);
	target.write(bytes);
}

/**
 * Maybe not copied, if want copy, please use copyTo.
 */
public static byte[] getBytes(MemorySegment[] segments, int baseOffset, int sizeInBytes) {
	// avoid copy if `base` is `byte[]`
	if (segments.length == 1) {
		byte[] heapMemory = segments[0].getHeapMemory();
		// 基于byte[]数组的memorysegment
		if (baseOffset == 0
				&& heapMemory != null
				&& heapMemory.length == sizeInBytes) {
			return heapMemory;
		} else {
			// 将内存从堆外内存拷贝出来
			byte[] bytes = new byte[sizeInBytes];
			segments[0].get(baseOffset, bytes, 0, sizeInBytes);
			return bytes;
		}
	} else {
		byte[] bytes = new byte[sizeInBytes];
		BinaryRowUtil.copySlow(segments, baseOffset, bytes, 0, sizeInBytes);
		return bytes;
	}
}

BinaryString有什么好处呢？

  1. 仅仅序列化一次
  2. 是可以修改的string，而不会产生中间对象
  3. 序列化的时候仅仅是内存的拷贝

BinaryArray

同样的基于memorysegment实现的还有BinaryMap和BinaryArray，这两者都有一个Generic的实现用以快速的更新, GenericArray要求数据类型都是相同的类型

BinaryArray的存储格式:

[numElements(int)] + [null bits(4-byte word boundaries)] + [values or offset&length] + [variable length part].

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
public void serialize(BaseArray record, DataOutputView target) throws IOException {
	BinaryArray binaryArray = baseArrayToBinary(record);
	target.write(binaryArray.getBytes());
}

public BinaryArray baseArrayToBinary(BaseArray from) {
	if (from instanceof BinaryArray) {
		return (BinaryArray) from;
	}

	int numElements = from.numElements();
	if (reuseBinaryArray == null) {
		reuseBinaryArray = new BinaryArray();
	}
	if (reuseBinaryWriter == null || reuseBinaryWriter.getNumElements() != numElements) {
		reuseBinaryWriter = new BinaryArrayWriter(
			reuseBinaryArray, numElements, BinaryArray.calculateElementSize(eleType));
	} else {
		reuseBinaryWriter.reset();
	}

	for (int i = 0; i < numElements; i++) {
		if (from.isNullAt(i)) {
			reuseBinaryWriter.setNullAt(i, eleType);
		} else {
			BaseRowUtil.write(reuseBinaryWriter, i,
					TypeGetterSetters.get(from, i, eleType), eleType, elementSerializer);
		}
	}
	reuseBinaryWriter.complete();

	return reuseBinaryArray;
}

从BinaryArraySerializer可以看到，处理方式和BaseRow很像，先baseToBinary，然后直接从segments拷贝byte。

DataStructureConverters

以上的类型和这个类型convert搭配使用才发挥出相应的效果，这个工具类的作用在于在codegen的阶段，根据输入的类型去转化为相对应的InternalType

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def genToInternal(ctx: CodeGeneratorContext, t: DataType): String => String = {
    val iTerm = boxedTypeTermForType(t.toInternalType)
    val eTerm = externalBoxedTermForType(t)
    if (isIdentity(t)) {
      term => s"($iTerm) $term"
    } else {
      val scalarFuncTerm = classOf[BuildInScalarFunctions].getCanonicalName
      TypeConverters.createExternalTypeInfoFromDataType(t) match {
        case Types.STRING => term => s"$BINARY_STRING.fromString($term)"
        case Types.SQL_DATE | Types.SQL_TIME =>
          term => s"$scalarFuncTerm.safeToInt(($eTerm) $term)"
        case Types.SQL_TIMESTAMP => term => s"$scalarFuncTerm.safeToLong(($eTerm) $term)"
        case _ =>
          val converter = genConvertField(ctx, createToInternalConverter(t))
          term => s"($iTerm) $converter.apply($term)"
      }
    }
  }
  

// 生成的类型转化函数
val internal = genToInternalIfNeeded(ctx, resultExternalType, resultClass, javaTerm)
        s"""
            |$javaTypeTerm $javaTerm = ($javaTypeTerm) $evalResult;
            |$resultTerm = $javaTerm == null ? null : ($internal);
            """.stripMargin

这样在codegen阶段就完成了相应类型的替换

BinaryRow和BinaryArray的底层存储

上面我们看到了binaryRow的使用方式，通过writeXXX的方式将数据写入到一个row中，一个row中可以写入基本类型，也可以写入binaryString, binaryArray，binaryMap等等变长的数据结构，其存储方式如下所示：

BinaryRow

A Row has two part: Fixed-length part and variable-length part.
Fixed-length part contains null bit set and field values. Null bit set is used for null tracking and is aligned to 8-byte word boundaries. Field values holds fixed-length primitive types and variable-length values which can be stored in 8 bytes inside. If it do not fit the variable-length field, then store the length and offset of variable-length part. Fixed-length part will certainly fall into a MemorySegment, which will speed up the read and write of field.
Variable-length part may fall into multiple MemorySegments.

在将其他格式通过baseRowToBinaryRow的时候，确定了BinaryRow包含的field个数，

1
2
3
4
5
6
7
8
9
public BinaryRowWriter(BinaryRow row, int initialSize) {
	this.nullBitsSizeInBytes = BinaryRow.calculateBitSetWidthInBytes(row.getArity());
	this.fixedSize = row.getFixedLengthPartSize();
	this.cursor = fixedSize;

	this.segment = MemorySegmentFactory.wrap(new byte[fixedSize + initialSize]);
	this.row = row;
	this.row.pointTo(segment, 0, segment.size());
}
1
2
3
4
public static int calculateBitSetWidthInBytes(int arity) {
	// add 8 bit header
	return ((arity + 63 + 8) / 64) * 8;
}

这个函数是计算出null值得Flag位需要多少Byte来表示，这里的+63是将其对其到8的倍数（向上去整的意思），+8和spark代码相比其实是因为flink多了一个header存储回撤消息的标志位,null bits中第一个byte存储了header位的信息，这里的null bits的作用主要是用在runtime处理时可以快速判断一条数据是不是null值。

1
2
3
4
5
6
7
8
9
public boolean anyNull() {
	// 这里有一个疑问，判断null的时候不是应该跳过第一个header位吗
	for (int i = 0; i < nullBitsSizeInBytes; i += 8) {
		if (segment.getLong(i) != 0) {
			return true;
		}
	}
	return false;
}
1
2
3
public int getFixedLengthPartSize() {
	return nullBitsSizeInBytes + 8 * arity;
}

获取整个固定长度字段的长度，再写变长区时就从这个offset写起。

写定长数据

1
2
3
public void writeByte(int pos, byte value) {
	segment.put(getFieldOffset(pos), value);
}

写不定长的数据

1
2
3
4
5
6
7
8
9
10
public void writeString(int pos, String input) {
	byte[] bytes = StringUtf8Utils.allocateBytes(input.length() * MAX_BYTES_PER_CHAR);
	int len = StringUtf8Utils.encodeUTF8(input, bytes);
	if (len <= 7) {
		// 小于记录length的长度时直接写在固定长度区
		writeLittleBytes(segment, getFieldOffset(pos), bytes, len);
	} else {
		writeBigBytes(pos, bytes, len);
	}
}

BinaryRow/BinaryRowWriter的实现和Spark中UnsafeRow/UnsafeRowWriter的实现非常相似，spark中的对此数据结构解释更为清晰一些。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/**
 * An Unsafe implementation of Row which is backed by raw memory instead of Java objects.
 *
 * Each tuple has three parts: [null bit set] [values] [variable length portion]
 *
 * The bit set is used for null tracking and is aligned to 8-byte word boundaries.  It stores
 * one bit per field.
 *
 * In the `values` region, we store one 8-byte word per field. For fields that hold fixed-length
 * primitive types, such as long, double, or int, we store the value directly in the word. For
 * fields with non-primitive or variable-length values, we store a relative offset (w.r.t. the
 * base address of the row) that points to the beginning of the variable-length field, and length
 * (they are combined into a long).
 *
 */

题外话

看代码的时候遵循的是一种推理加源码追踪的手段，但是代码在初始设计的时候应该是另一种维度的思考，因此应该换一种思路去想如果自己来实现这个feature需要考虑什么地方，多多思考。

谢谢支持