1. Background
PolarDB Distributed Edition (PolarDB for Xscale, abbreviated as: PolarDB-X) is a cloud-native distributed database with important features such as online transaction and analytical processing (HTAP), separation of computing and storage, and global secondary indexing. In terms of HTAP, PolarDB Distributed Edition has explored and practiced many aspects of vectorization for the AP engine, such as implementing columnar memory layout, MPP, and column-oriented executors, among other advanced features (refer to “PolarDB-X Vectorized Execution Engine”[1] and “PolarDB-X Vectorized Engine”[2]).
PolarDB Distributed Edition is fully developing a columnar storage node, responsible for providing columnar storage data, and is building a new HTAP architecture combined with row-column mixed storage + distributed computing nodes. It will officially launch on public cloud soon, and will also release open source in the future. Moreover, the most typical scenario for column storage is vectorization, where SIMD instructions, as a key part of vectorization, have been used by many mainstream AP engines to enhance computational speed. However, due to the limitations of the Java language, the CN computing engine of PolarDB Distributed Edition cannot actively call SIMD instructions before JDK 17, and in JDK 17, the official Java provides a wrapper for SIMD instructions, namely the Vector API.This article will introduce PolarDB Distributed Edition’s exploration and practice of vectorized SIMD instructions, including basic usage and implementation principles, as well as thoughts and insights on specific operator implementations.2. Introduction to SIMDSIMD (Single Instruction Multiple Data) is a type of processor instruction, where a single instruction can process multiple data simultaneously. Below are the scalar (Scalar) and SIMD (Vector) execution methods for addition:
To support SIMD programming, CPUs provide a series of special registers and instructions:
● Registers:
128-bit registers XMM0-XMM15 in the SSE instruction set;
256-bit registers YMM0-YMM15 in the AVX instruction set;● Arithmetic operations: PADDB, which calculates the sum of two groups of 8-bit integers, each group contains 16 8-bit integers (SSE2), can be used for both unsigned and signed types;● Comparison operations: PCMPEQB, which compares two groups of 8 bits for equality, each group contains 16 8 bits (SSE2), 32 (AVX2), 64 (AVX512BW);● Bit operations: PAND: performs a bitwise AND operation on the values of two registers; POR: same as PAND, OR operation; PXOR: same as PAND, XOR operation;
● Load/Store instructions
MOVAPS: moves a 128-bit value each time.
3. Introduction to Vector API
Before JDK 17, Java could not actively call SIMD instructions, but in JDK 17, the official Java provided a wrapper for SIMD instructions – the Vector API. The Vector API provides registers such as IntVector, LongVector, which automatically select the appropriate instruction set based on the underlying CPU, allowing developers to quickly perform SIMD programming without worrying about the specific CPU architecture. It also provides operations such as add, sub, xor, or for SIMD calculations, as well as fromArray, intoArray to read multiple data at once.
Basic Usage of Vector API
We will quickly introduce Vector API with an example of adding an array. In the Vector API, each Vector represents a register that can hold several elements, depending on the size of the register and the type of elements. For example, when the register size is 128 bits, it can hold 4 int types (each int occupies 32 bits)
public class LongSumBenchmark { // Define SPECIES to indicate the type of Vector private static final VectorSpecies<Long> SPECIES = LongVector.SPECIES_PREFERRED; private int count; private long[] longArr; private long[] longArr2; private long[] longArr3; public void normalSum() { for (int i = 0; i < longArr.length; i++) { longArr3[i] = longArr[i] + longArr2[i]; } } public void vectorSum() { int i; int batchSize = longArr.length; int end = SPECIES.loopBound(batchSize); // Get the aligned upper limit through loopBound for (i = 0; i < end; i += SPECIES.length()) { // fromArray(SPECIES, longArr, i) indicates to extract SPECIES.length() elements from longArr starting from the i-th position LongVector va = LongVector.fromArray(SPECIES, longArr, i); LongVector vb = LongVector.fromArray(SPECIES, longArr2, i); LongVector vc = va.add(vb); // Call add function to sum using SIMD instructions // intoArray(longArr3, i) indicates to store the contents of the vc register into longArr3 starting from the i offset vc.intoArray(longArr3, i); } for(; i < batchSize; ++i) { // The remaining part needs manual processing longArr3[i] = (longArr[i] + longArr2[i]); } }}FMA Calculation
To demonstrate the performance improvement of Vector API, we reproduced an example of FMA calculation.
FMA addition refers to: c = c + a[i] * b[i]. Where a and b are both float/double type arrays:
@Benchmarkpublic double normalSum() { double sum = 0; for (int i = 0; i < doubleArr.length; i++) { sum += doubleArr[i] * doubleArr2[i]; } return sum;}@Benchmarkpublic double vectorSum() { var sum = DoubleVector.zero(SPECIES); int i; int batchSize = doubleArr.length; var upperBound = SPECIES.loopBound(doubleArr.length); for (i = 0; i < upperBound; i += SPECIES.length()) { DoubleVector va = DoubleVector.fromArray(SPECIES, doubleArr, i); DoubleVector vb = DoubleVector.fromArray(SPECIES, doubleArr2, i); sum = va.fma(vb, sum); } var c = sum.reduceLanes(VectorOperators.ADD); for(; i < batchSize; ++i) { doubleArr3[i] = (doubleArr[i] + doubleArr2[i]); } return c;}Testing environment: Randomly generated 1000/100000 double precision floating-point numbers.Testing results: Vectorized execution is twice as fast as scalar execution.
Result Analysis:
● Execution result of Vector API: Only one instructionvfmadd231pd %ymm0,%ymm2,%ymm3: Multiply the double precision floating point numbers in ymm2 and ymm3, add them to the data in ymm1, and put the result in ymm1
0x00007f764d363902: vfmadd231pd %ymm0,%ymm2,%ymm3● Result of scalar execution: The multiplication and addition are split into two instructions vmulsd and vaddsd
0x00007f000133050d: vmulsd 0x10(%rax,%r13,8),%xmm0,%xmm00x00007f0001330514: vaddsd %xmm0,%xmm1,%xmm1From the test results, it can be seen that for FMA calculation scenarios, the Vector API merges the vmulsd and vaddsd that originally required two instructions into one instruction vfmadd.
However, it should be noted that the optimization of FMA calculations cannot be applied in databases because PolarDB-X executes multiplication and addition as two separate operators.
Using Vector API to Implement Basic SIMD Operations
Here we will demonstrate how to use Vector API to implement the Gather and Scatter operations in the paper Rethinking SIMD Vectorization for In-Memory Databases1. Gather: The fromArray operation of Vector API provides a wrapper for Gather operations, we just need to pass in the corresponding parameters.
a. Scalar implementation
public void gather(int[] source, int[] indexes, int count, int[] target) { for (int i = 0; i < count; i++) { target[i] = source[indexes[i]]; }}b. SIMD implementation
public void gather(int[] source, int[] indexes, int count, int[] target) { final int laneSize = INTEGER_VECTOR_SPECIES.length(); final int indexVectorLimit = count / laneSize * laneSize; int indexPos = 0 for (; indexPos < indexVectorLimit; indexPos += laneSize) { IntVector av = IntVector.fromArray(INTEGER_VECTOR_SPECIES, source, 0, indexes, indexPos); av.intoArray(target, indexPos); } if (indexPos < indexLimit) { scalarPrimitives.gather(source, indexes, indexPos, indexLimit - indexPos, target, targetPos); }}2. Scatter implementation: Similarly, the intoArray operation of Vector provides a wrapper for Scatter operations.
a. Scalar implementation
public void scatter(long[] source, long[] target, int[] scatterMap, int copySize) { for (int i = 0; i < copySize; i++) { target[scatterMap[i]] = source[i]; }}b. SIMD implementation
public void scatter(int[] source, int[] target, int[] scatterMap, int copySize) { int laneSize = SIMDHandles.INT_VECTOR_LANE_SIZE; // Number of bits that SIMD can process at a time final int indexVectorLimit = copySize / laneSize * laneSize; int index = 0; for (; index < indexVectorLimit; index += laneSize) { IntVector dataInVector = IntVector.fromArray(INTEGER_VECTOR_SPECIES, source, index); // Extract K numbers from source[index] dataInVector.intoArray(target, 0, scatterMap, index); } if (index < copySize) { scalarPrimitives.scatter(source, target, scatterMap, index, copySize - index); }}Vector API Implementation Principles
Taking vectorized addition as an example, let’s explore the implementation of the add function:
LongVector va = LongVector.fromArray(SPECIES, longArr, i);LongVector vb = LongVector.fromArray(SPECIES, longArr2, i);LongVector vc = va.add(vb);At the JDK Level
At the JDK level, Java does not perform any optimization; its underlying implementation simply calls the apply function for each element in the Vector, and the apply function points to a bound function whose implementation is scalar addition. Obviously, this approach may even increase the execution overhead, so where does the high performance of the Vector API come from?
1. Implementation of the add function
@Override@ForceInlinepublic final LongVector add(Vector<Long> v) { return lanewise(ADD, v);}2. Finally, it calls the b0pTemplate function for computation
@ForceInlinefinalLongVector bOpTemplate(Vector<Long> o, FBinOp f) { long[] res = new long[length()]; long[] vec1 = this.vec(); long[] vec2 = ((LongVector)o).vec(); for (int i = 0; i < res.length; i++) { res[i] = f.apply(i, vec1[i], vec2[i]); } return vectorFactory(res);}3. apply binds to the function implementation of scalar execution
LongVector lanewiseTemplate(VectorOperators.Binary op, Vector<Long> v) { LongVector that = (LongVector) v; that.check(this); .... int opc = opCode(op); return VectorSupport.binaryOp( opc, getClass(), long.class, length(), this, that, BIN_IMPL.find(op, opc, (opc_) -> { switch (opc_) { case VECTOR_OP_ADD: return (v0, v1) -> v0.bOp(v1, (i, a, b) -> (long)(a + b)); case VECTOR_OP_SUB: return (v0, v1) -> v0.bOp(v1, (i, a, b) -> (long)(a - b)); case VECTOR_OP_MUL: return (v0, v1) -> v0.bOp(v1, (i, a, b) -> (long)(a * b)); case VECTOR_OP_DIV: return (v0, v1) -> v0.bOp(v1, (i, a, b) -> (long)(a / b)); case VECTOR_OP_ADD: return (v0, v1) -> v0.bOp(v1, (i, a, b) -> (long)(a + b)); .... }}));}At the JVM Level
1. Preliminary Knowledge: JIT and C2 Compiler
Here we need to briefly explain Java’s Just-In-Time compilation (JIT). The overall execution process of Java can be divided into interpreted execution and compiled execution (JIT). The first step is to compile the source code into bytecode using javac and interpret it. During interpreted execution, the JVM collects the running information of the program. For the hotspot code among them, the compiler (default is C2) compiles it, converting bytecode directly into machine code, and then executing it.How is it determined to be hotspot code? The JVM sets a threshold, and when the number of calls to a method or code block exceeds this threshold within a certain time, it is considered hotspot code.The execution process of JIT is as follows:
2. Implementation of Vector API at the JVM Level
Through research, we found the first commit of the Vector API. From this commit, we discovered that there are many modifications to the JVM core in the implementation of the Vector API. First, it adds Vector API related intrinsics in the c2compiler (src/hotspot/share/opto/c2compiler.cpp)
When JIT is triggered, the Vector API related code will be replaced with JVM level intrinsics, optimized using SIMD instructions. Specifically, when Vector API code is JIT compiled, it replaces the original IR nodes in the syntax tree with the implementation of intrinsics, that is, replacing method with the intrinsic method.
//---------------------------make_vm_intrinsic----------------------------CallGenerator* Compile::make_vm_intrinsic(ciMethod* m, bool is_virtual) { vmIntrinsicID id = m->intrinsic_id(); C2Compiler* compiler = (C2Compiler*)CompileBroker::compiler(CompLevel_full_optimization); bool is_available = false; methodHandle mh(THREAD, m->get_Method()); is_available = compiler != NULL && compiler->is_intrinsic_supported(mh, is_virtual) && !C->directive()->is_intrinsic_disabled(mh) && !vmIntrinsics::is_disabled_by_flags(mh); if (is_available) { return new LibraryIntrinsic(m, is_virtual, vmIntrinsics::predicates_needed(id), vmIntrinsics::does_virtual_dispatch(id), id); } else { return NULL; }}The final SIMD instructions are generated in the C2 assembler
void Assembler::addpd(XMMRegister dst, XMMRegister src) { NOT_LP64(assert(VM_Version::supports_sse2(), "")); InstructionAttr attributes(AVX_128bit, /* rex_w */ VM_Version::supports_evex(), /* legacy_mode */ false, /* no_mask_reg */ true, /* uses_vl */ true); attributes.set_rex_vex_w_reverted(); int encode = simd_prefix_and_encode(dst, dst, src, VEX_SIMD_66, VEX_OPCODE_0F, &attributes); emit_int16(0x58, (0xC0 | encode));}4. Expression Calculation
Adding Long Arrays
The greatest advantage of Vector API is to accelerate calculations, so next we will explore scenarios where it can bring performance improvements. First, we conducted a benchmark on the long array addition scenario given earlier. It can be seen that in the array addition scenario, scalar execution and SIMD execution are not much different. By tracking the assembly instructions, we found that whether SIMD execution or scalar execution ultimately generates the vpaddq instruction:
1. SIMD execution
0x00007f3eb13602ae: vpaddq 0x10(%r11,%rbx,8),%ymm3,%ymm3;2. Scalar execution
0x00007fa6fd33259a: vpaddq 0x10(%r8,%rbp,8),%ymm4,%ymm4vpaddq: AVX512 instruction. Adds the 16 bytes of data starting from (%r11 + %rbx * 8) with the data in the ymm3 register, writing to the ymm3 register.
Auto-Vectorization
We found that even without explicitly using the Vector API, the vectorized addition code would still be vectorized, thanks to Java’s auto-vectorization mechanism. The implementation of Java’s auto-vectorization is similar to the Vector API; it checks during JIT compilation whether the code can use SIMD instructions for calculations, and if so, replaces it with SIMD implementations. However, auto-vectorization can only handle relatively simple calculations, while complex calculations still require manual SIMD (using the Vector API). For example, known limitations of auto-vectorization include:
1. Only supports incrementing for loops
2. Only supports Int/Long types (Short/Byte/Char indirectly supported through int)
3. The upper limit of the loop must be a constantThrough research on some mature OLAP systems (such as ClickHouse) and exploration of our actual scenarios, we used the Vector API to reimplement a batch of operators, showcasing four representative scenarios.5. Case Conversion
Using SIMD to implement case conversion is relatively simple; we just need to call the compare method for comparison and use the lanewise method for XOR operation. Here we introduce the VectorMask, which can be understood as a boolean type register containing n 0/1.
@Benchmarkpublic void normal() { final long Mask = 'A' ^ 'a'; for (int i = 0; i < charArr.length; i++) { if(charArr[i] >= 'A' && charArr[i] <= 'Z') { charArr[i] ^= Mask; } }}@Benchmarkpublic void vector() { final long Mask = 'A' ^ 'a'; int i; int batchSize = charArr.length; int end = SPECIES.loopBound(charArr.length); for (i = 0; i < end; i += SPECIES.length()) { ByteVector from = ByteVector.fromArray(SPECIES, charArr, i); VectorMask mask = from.compare(VectorOperators.GE, 'A', from.compare(VectorOperators.LE, 'Z')); ByteVector to = from.lanewise(VectorOperators.XOR, Mask, mask); // Here we pass in the mask, indicating that XOR should only be executed at positions where mask=1 to.intoArray(charArr, i); } for(; i < batchSize; ++i) { if(charArr[i] >= 'A' && charArr[i] <= 'Z') { charArr[i] ^= Mask; } }}Benchmark
Testing environment: Randomly generated 1000 and 100000 byte type letters for benchmarking
Testing results: In the scenario with count=100000, it is 50 times faster, due to the length of ByteSpecies being 32, and the SIMD execution method avoiding the overhead of branch prediction failure flush pipelines.
6. SIMD FilterUsing SIMD instructions to implement Filter can use Gather operations to read elements from the array and use the compare method for comparison, finally using bitwise operations to record the indices that meet the conditions.
@Benchmarkpublic void normal() { newSize = 0; for (int i = 0; i < intArr.length; i++) { if(intArr[sel[i]] <= intArr2[sel[i]]) { sel[newSize++] = i; } }}@Benchmarkpublic void vector() { newSize = 0; int i; int batchSize = intArr.length; int end = SPECIES.loopBound(intArr.length); for (i = 0; i < end; i += SPECIES.length()) { IntVector va = IntVector.fromArray(SPECIES, intArr, 0, sel, i); IntVector vb = IntVector.fromArray(SPECIES, intArr2, 0, sel, i); VectorMask<Integer> vc = va.compare(VectorOperators.LE, vb); if(!vc.anyTrue()) { continue; } int res = (int) vc.toLong(); while(res != 0) { int last = res && -res; // Find the last power of 2 corresponding to the last 1 in the binary, for example, 12 = 1100, 12 && (-12) = 4 sel[newSize++] = i + Integer.numberOfTrailingZeros(last); // numberOfTrailingZeros(2^i) = i res ^= last; } } for(; i < batchSize; ++i) { int j = sel[i]; if(intArr[j] <= intArr2[j]) { sel[newSize++] = i; } }}Benchmark
Testing environment: Randomly generated 1000 and 100000 int type integers for benchmarkingResult analysis: When count=1000, the performance of SIMD actually decreased, because at this time the function was not JIT compiled but used the scalar execution of the JDK level. When count=100000, the method had already been JIT compiled, thus the performance improved by 25%.
7. Local Exchange
What is the Local Exchange operator?
The Exchange operator is an important component for data shuffling in the MPP execution mode of PolarDB Distributed Edition. For more background, refer to the article “PolarDB-X Parallel Computing Framework”[3], which will not be elaborated here. Here, we mainly optimize the execution mode of LocalPartitionExchanger in the Local Exchange operator. In this practice, we achieved a 35% performance improvement by using vectorized operators + SIMD Scatter instructions.LocalPartitionExchanger can be simply understood as: For a driver in a downstream pipeline, it calculates the corresponding partition for each row of data in the Chunk and reassembles the data of the same partition into a new Chunk to feed to the upstream driver.
If you do not understand this statement, it is okay; the focus of this article is to introduce how to optimize code logic using SIMD instructions, so feel free to continue reading.
Non-Vectorized Version (Current Version of PolarDB Distributed Edition)
The current version of PolarDB Distributed Edition follows the Local Exchange operator under the row storage execution model. Its basic idea is to enumerate and calculate the partition for each row one by one, writing to the corresponding Chunk. The disadvantage of this execution mode is that it cannot effectively utilize columnar memory layout, and the appendTo operation incurs a significant time overhead.
The detailed execution process of the old version is as follows:
1. Calculate the partition corresponding to the position (row number): It uses n linked lists to save the position list corresponding to each partition
a. (partitionGenerator.getPartition can be simply understood as performing a hash operation on the data corresponding to the position to obtain the target position of the partition)
for (int position = 0; position < keyChunk.getPositionCount(); position++) { int partition = partitionGenerator.getPartition(keyChunk, position); partitionAssignments[partition].add(position);}2. buildChunk: Generate the corresponding partition Chunk
a. Enumerate the partition first
i. Enumerate the positions corresponding to the partition (position represents row)
1. Enumerate the column block corresponding to that row
a. Call the builder’s appendTo to add elements to the ChunkBuilder
b. appendTo incurs a virtual function call, which has a significant performance overhead!
Map<Integer, Chunk> partitionChunks = new HashMap<>();for (int partition = 0; partition < executors.size(); partition++) { // Enumerate partition List<Integer> positions = partitionAssignments[partition]; // Get the corresponding positions = partitionAssignments[partition] ChunkBuilder builder = new ChunkBuilder(types, positions.size(), context); Chunk partitionedChunk; for (Integer pos : positions) { // Enumerate the corresponding position (row) for (int i = 0; i < chunk.getBlockCount(); i++) { // Enumerate the corresponding column builder.appendTo(chunk.getBlock(i), i, pos); // Add the element at pos row and i column position to the corresponding ChunkBuilder } } partitionedChunk = builder.build(); partitionChunks.put(partition, partitionedChunk);}Local Exchange SIMD Optimization
Idea:
1. Optimize appendTo: The appendTo operation incurs significant overhead; try to use a more efficient method to copy data, such as system.arrayCopy
2. Optimize the row enumeration mode: The row-by-row enumeration method is limited because the data of each column is in different Blocks, making it impossible to batch copy. Additionally, the row-by-row enumeration method is not memory access friendly. Therefore, consider enumerating columns first.
● Question: Each partition’s corresponding rows are not continuous;
● Solution: As shown in the figure below, we hope to calculate the positionMapping array so that the rows of the same partition can be migrated to continuous rows, allowing us to use Scatter operations for acceleration.
afterValue[positionMapping[i]] = preValue[i]
3. Question: How to find positionMapping?● Actually quite simple; perform a linear scan to record the size of each partition, denoted as partitoinSize[]● Knowing the size allows us to calculate the offset of each partition in the final array, partitionOffset[]
● Knowing the offset allows us to calculate positionMapping while traversing: just record the offset corresponding to the partition of that row, and increment the offset. This process will be more intuitive with code.
Specific Steps1. Calculate positionMapping[]● Calculate the corresponding partition for each position and the size of each partition
private void scatterAssignmentNoneBucketPageInBatch(Chunk keyChunk, int positionStart, int batchSize) { int[] partitions = scatterMemoryContext.getPartitionsBuffer(); for (int i = 0; i < batchSize; i++) { int partition = partitionGenerator.getPartition(keyChunk, i + positionStart); partitions[i] = partition; ++paritionSize[partition]; // Count the size of each partition }}● Calculate the offset corresponding to each partition
private void prepareDestScatterMapStartOffset() { int startOffset = 0; for (int i = 0; i < partitionNum; i++) { partitionOffset[i] = startOffset; // Count the offset corresponding to each partition startOffset += partitionSize[i]; }}● Calculate positionMapping
protected void prepareLengthsPerPartitionInBatch(int batchSize, int[] partitions, int[] destScatterMap) { for (int i = 0; i < batchSize; ++i) { positionMapping[i] = partitionOffset[partitions[i]]++; // Get the memory address corresponding to each position after scatter }}2. Use scatter operations
● The vectorizedSIMDScatterAppendTo will check if the Block supports SIMD execution and call the Block’s copyPositions_scatter_simd method. Here is an example with IntBlock:
protected void scatterCopyNoneBucketPageInBatch(Chunk keyChunk, int positionStart, int batchSize) { Block[] sourceBlocks = keyChunk.getBlocksDirectly(); // Get all blocks for (int col = 0; col < types.size(); col++) { // Enumerate all columns Block sourceBlock = sourceBlocks[col]; // Get the block for (int i = 0; i < chunkBuilders.length; i++) { blockBuilders[i] = chunkBuilders[i].getBlockBuilder(col); // Get the pageBuilder for all partitions } vectorizedSIMDScatterAppendTo(sourceBlock, scatterMemoryContext, blockBuilders); }}public void copyPositions_scatter_simd(ScatterMemoryContext scatterMemoryContext, BlockBuilder[] blockBuilders) { // Buffer to store destination data int[] afterValue = scatterMemoryContext.getafterValue(); // Buffer to store original data int[] preValue = scatterMemoryContext.getpreValue(); // positionMapping int[] positionMapping = scatterMemoryContext.getpositionMapping(); // Size of each partition int[] partitionSize = scatterMemoryContext.getpartitionSize(); // Use scatter operation VectorizedPrimitives.SIMD_PRIMITIVES_HANDLER.scatter(preValue, afterValue, positionMapping, 0, batchSize); int start = 0; for (int partition = 0; partition < blockBuilders.length; partition++) { IntegerBlockBuilder unsafeBlockBuilder = (IntegerBlockBuilder) blockBuilders[partition]; // Note that the implementation of writeBatchInts here unsafeBlockBuilder.writeBatchInts(afterValue, bufferNullsBuffer, start, partitionSize[partition]); start += partitionSize[partition]; }}3. Use writeBatchInts to implement memory copying
public IntegerBlockBuilder writeBatchInts(int[] ints, boolean[] nulls, int sourceIndex, int count) { values.addElements(getPositionCount(), ints, sourceIndex, count); valueIsNull.addElements(getPositionCount(), nulls, sourceIndex, count); return this;}addElements uses the System.arraycopy command for batch memory copying. System.arraycopy is a built-in function of the JVM, and its implementation is much more efficient than calling the append interface to add data one by one.
@IntrinsicCandidatepublic static native void arraycopy(Object src, int srcPos, Object dest, int destPos, int length);Benchmark
Testing environment: We set the number of partitions to 3 and input 100 chunks of size 1024, each containing 4 columns of int to the Local Exchange operator for benchmarking.
Result Analysis:
We found that purely vectorized algorithms do not yield performance improvements. The reason is that the bottleneck of the Local Exchange operator lies in the appendTo operation. However, the SIMD Scatter + System.arrayCopy method achieved a 35% performance improvement.
8. SIMD Hash Join
SIMD Idea
In CMU 15-721, the implementation of SIMD Hash Probe is mentioned, and we will use the Vector API to reproduce this process.
First, let’s review the process of SIMD Hash Probe using open addressing. The basic idea is to probe four positions’ elements at once.
1. Calculate the hash values for four positions
2. Use Gather operations to read the elements at the corresponding positions in the hash table
3. Use SIMD’s compare operation to check for hash conflicts
4. The first three steps are understandable; next, we need to let elements with hash conflicts address the next position to continue resolving hash conflicts. This step can be implemented using SIMD addition. The key point is how to move the elements without conflicts away and bring in the elements from the back of the array to continue matching?(For example, we hope to replace k1, k4 in the figure below with k5, k6)
Solution:
a. Use the expand operation to perform selective load operations
buildPositionVec = buildPositionVec.expand(probePosition, probeOffset, probeMask);b. expand operation can replace the n position elements that are 1 in the probeMask with n elements from the hashPosition array starting from probeOffset.
1. For example, in the above case, the data changes as follows2. hashPosition = [h1, h2, h3, h4, h5, h6, h7, h8…..]3. probeOffset = 44. probeMask = [1, 0, 0, 1]5. Input: [h1, h2, h3, h4]6. Output: [h5, h2, h3, h6]
Side Note: How to Implement a More Powerful Expand
In the code above, our expand operation takes three parameters, but the official expand function of openJDK can only take one parameter (the expand usage of openJDK cannot pass arrays, it can only expand between Vectors)
This was made possible by Alibaba Cloud’s strong self-research capabilities. We actively communicated with the Alibaba Cloud JVM team, and the team helped us implement a richer expand operation.
Verification
The Hash method of PolarDB Distributed Edition is not open addressing, but cuckoo Hash; however, the SIMD principle is similar. Here we provide the process of modifying the SIMD Hash Probe for cuckoo Hash.
@Benchmarkpublic void normal() { for(int i = 0; i < count; i++) { int joinedPosition = 0; int matchedPosition = (int) hashTable.keys[hashedProbeKey[i]]; while (matchedPosition != LIST_END) { if (buildKey[matchedPosition] == probeKey[i]) { joinedPosition = matchedPosition; break; } matchedPosition = (int) positionLinks[matchedPosition]; } joinedPositions[i] = joinedPosition; }}@Benchmarkpublic void vector() { int probeOffset = 0; VectorMask<Long> probeMask; VectorMask<Integer> intProbeMask; LongVector probeValueVec = LongVector.fromArray(LongVector.SPECIES_512, probeKey, probeOffset); // step 1: Use Gather operation to read data from hashTable's keys. int matchedPosition = hashTable.keys[hashedProbeKey[i]]; LongVector buildPositionVec = LongVector.fromArray(LongVector.SPECIES_512, hashTable.keys, 0, hashedProbeKey, probeOffset); IntVector probeIndexVec = IntVector.fromArray(IntVector.SPECIES_256, index, probeOffset); probeOffset += 8; while(probeOffset + LongVector.SPECIES_512.length() < count){ // step 2: Calculate the positions in buildPositionVec that are 0 and handle those separately VectorMask<Long> emptyMask = buildPositionVec.compare(VectorOperators.EQ, 0); // step 3: Use Gather operation to calculate the values of buildKey[matchedPosition[i]] IntVector intbuildPositionVec = buildPositionVec.castShape(IntVector.SPECIES_256, 0).reinterpretAsInts(); LongVector buildValueVec = LongVector.fromArray(LongVector.SPECIES_512, intbuildPositionVec, buildKey); // step 4: Use SIMD compare to perform comparison. if (buildKey[matchedPosition] == probeKey[i]) VectorMask<Long> valueEQMask = probeValueVec.compare(VectorOperators.EQ, buildValueVec, emptyMask.not()); // step 5: Calculate the mask of successful probe to find the target position probeMask = valueEQMask.or(emptyMask); intProbeMask = probeMask.cast(IntVector.SPECIES_256); // step 6: Use scatter operation to write the matched position into joinedPosition. joinedPositions[i] = matchedPosition; buildPositionVec.intoArray(joinedPositions, probeIndexVec, valueEQMask); // step 7: Handle hash conflicts. matchedPosition = positionLinks[matchedPosition]; buildPositionVec = LongVector.fromArray(LongVector.SPECIES_512, intbuildPositionVec, positionLinks); // step 8: Use expand operation to get the next probe vector buildPositionVec = buildPositionVec.expand(probePosition, probeOffset, probeMask); probeValueVec = probeValueVec.expand(probeKey, probeOffset, probeMask); probeIndexVec = probeIndexVec.expand(index, probeOffset, intProbeMask); // step 9: Update probeOffset probeOffset += probeMask.trueCount(); } scalarProcessVector(probeValueVec, buildPositionVec, probeIndexVec, joinedPositions); // Process the last Vector if (probeOffset < count) { processBatchX(hashedProbeKey, probeOffset, count, joinedPositions); }}Benchmark
Testing environment: During the Build phase, we inserted 1 million elements with sizes ranging from 0 to 1000 into the hash table to simulate hash conflicts, and during the Probe phase, we calculated the time required to probe 1 million elements.
Testing results: Although we implemented SIMD Hash Probe, due to the existing Vector API’s insufficient support for types (the code contains a lot of type conversions), the actual results were not excellent, even resulting in a performance drop of more than 2 times.
However, setting aside the performance drop caused by the Java Vector API itself, using SIMD instructions to optimize Hash Probe may not be a wise move, as the bottleneck of Hash Join does not lie in computation, but in memory access. “Improving hash join performance through prefetching” states that 73% of the overhead in the database’s Hash Join algorithm comes from CPU cache misses, which also explains why SIMD instructions did not optimize it. In internal testing of PolarDB Distributed Edition on TPC-H Q9, it was found that CPU cycles wasted on cache misses reached 10 times that of computation, so we shifted our optimization of Hash Join to cache misses. PolarDB-X has already tried to use prefetch instructions for prefetching (enhanced by the Alibaba Cloud JVM team) and vectorized Hash Probe to optimize cache misses, and related optimization results will be showcased in subsequent articles.9. SummaryIn this article, we first introduced the usage and implementation principles of the Vector API, focusing on its application in database scenarios. We achieved a 50-fold performance improvement in case conversion, a 25% improvement in the Filter operator, and a 35% improvement in the Local Exchange operator. We also discussed the limitations of the Vector API in operators like Hash Probe, where cache misses are the bottleneck.As a distributed HTAP database, PolarDB Distributed Edition has always prioritized performance optimization of the AP engine. We not only focus on common optimizations in the industry but are also actively exploring the “deep water zone” of mixed storage architecture and vectorized SIMD instructions. Stay tuned for the official release of the columnar storage engine of PolarDB Distributed Edition on public cloud and open source.
References:
[1] PolarDB-X Vectorized Execution Engine: https://zhuanlan.zhihu.com/p/337574939
[2] PolarDB-X Vectorized Engine: https://zhuanlan.zhihu.com/p/339514444
[3] PolarDB-X Parallel Computing Framework: https://zhuanlan.zhihu.com/p/346320114
PolarDB Distributed Edition Big Price Drop 🎉
Cloud-native database PolarDB Distributed Edition (PolarDB-X) has undergone a significant price drop of40%, with prices starting at0.75 yuan/hour. Click “Read the original text” at the end of the article to see the details~