背景大数据业务一直是鲲鹏服务器的重要板块，本轮针对大数据离线引擎进行优化，该引擎基于meta开发的velox引擎改造而来。大数据引擎的整体框架及大部分算子、表达式都沿用了velox引擎代码，我们在性能分析、编写优化代码前都先在velox上分析，后续验证优化有效后，再将代码合入到大数据引擎，下面的文章我们将以velox为例进行分析。从tpcds各sql的火焰图来看，在算子层面，占比比较多的是tablescan、hashagg、hashsort、filter等算子；如下图q14a火焰图所示，tablescan在图中占比达5%；如果能在里面找到占比较大的可优化函数，效果还是很可观的，因此我们需要进一步分析一.优化思路出发点：tableScan算子是否存在优化点；分析路径：继续从q14a火焰图上分析，tableScan的热点主要集中在函数void readWithVisitor(const uint64_t* nulls, Visitor visitor)，而readWithVisitor内部存在fastPath（快路径）的处理方式。在快路径中，你可以使用向量化加速处理数据，但仅限于X86架构机器，其核心处理逻辑位于BitPackDecoder.cpp -> void unpack()中（待补函数调用树）。由于鲲鹏芯片是arm架构，我们可以采用sve指令改写原有逻辑。二.原代码 2.1 unpack函数所处文件：BitPackDecoder.cpp 函数逻辑：这个函数从位压缩（bit-packed）的数据流中解压出指定行的值。在列式存储格式（如ORC、Parquet）中，数值类型的数据经常按固定位宽压缩存储。示例：原始数据：[5, 3, 7, 2, 6, 1, 4, 0] 位压缩表示（每个值3位）：二进制：101 011 111 010 110 001 100 000 十六进制：0xB7 0xB1 0x80 (3字节) 存储布局（以小端序为例）：内存地址： 0x1000 0x1001 0x1002 二进制： 10101111 10101100 10000000 [5][3][7] [2][6][1] [4][0] 整个函数的执行逻辑可以分为以下几个部分： ●针对0位宽的数据流特例处理 ●调整bitOffset，针对位宽大于56的数据流进行特例处理 ●对1-24位宽的数据使用专门的AVX2实现，逻辑在decode1To24 ()中实现 ●针对其余情况使用标量处理路径因此，我们可以聚焦于方法decode1To24 ()。 2.2 decode1To24函数所处文件：BitPackDecoder.cpp 函数逻辑：一个高度优化的SIMD（AVX2）位解包函数，专门处理1-24位宽的位压缩数据。该函数最开始先计算需要的掩码，后续是一个主循环结构，每次处理8行数据，循环内主要分两个路径： 2.2.1.连续行优化：（当8行连续且位宽≤16时）里面会按需使用到loadBits、_pdep_u64、_mm_load_si128、_mm256_cvtepu16_epi32等AVX2指令；其中又有两条子路径，一个是位宽<=8，另一个是位宽9-16，此处以位宽<=8讲解 template int32_t decode1To24( const uint64_t* bits, int32_t bitOffset, const int* rows, int32_t numRows, T* result) { constexpr uint64_t kMask = bits::lowMask(width); constexpr uint64_t kMask2 = kMask | (kMask << 8); constexpr uint64_t kMask4 = kMask2 | (kMask2 << 16); constexpr uint64_t kDepMask8 = kMask4 | (kMask4 << 32); int32_t i = 0; for (; i + 8 <= numRows; i += 8) { auto row = rows[i]; auto endRow = rows[i + 7]; __m256i eightInts; if (width <= 16 && endRow - row == 7) { //当8行连续且位宽≤16 // Special cases for 8 contiguous values with <= 16 bits. if (width <= 8) { uint64_t eightBytes; if (width == 8) { if (!bitOffset) { eightBytes = *addBytes(bits, row); } else { eightBytes = bits::detail::loadBits (bits, bitOffset + 8 * row, 64); } } else { auto bit = row * width + bitOffset; auto byte = bit >> 3; auto shift = bit & 7; uint64_t word = *addBytes(bits, byte) >> shift; eightBytes = _pdep_u64(word, kDepMask8); } eightInts = _mm256_cvtepu8_epi32( _mm_loadl_epi64(reinterpret_cast (&eightBytes))); } else {...} //位宽9-16 } else {...} //非连续或位宽>16 store8Ints(eightInts, i, result); } } // loadBits()位于BitUtil.h // 当width == 8时，T为uint64_t（一次循环处理八个数据，每个数据位宽为8，刚好是64位）， template inline T loadBits(const uint64_t* source, uint64_t bitOffset, uint8_t numBits) { constexpr int32_t kBitSize = 8 * sizeof(T); auto address = reinterpret_cast (source) + bitOffset / 8; T word = *reinterpret_cast (address); auto bit = bitOffset & 7; if (!bit) { return word; } if (numBits + bit <= kBitSize) { return word >> bit; } uint8_t lastByte = reinterpret_cast (address)[sizeof(T)]; uint64_t lastBits = static_cast (lastByte) << (kBitSize - bit); return (word >> bit) | lastBits; } 当位宽为8，调用loadBits函数处理，获取经偏移后的uint64_t字节流eightBytes，里面存储了本轮循环要处理的八个字节；当位宽小于8，调用_pdep_u64指令配合kDepMask8掩码获取字节流eightBytes， _pdep_u64解析：PDEP指令一次将多个值解压到不同字节, 举例：假设width=3 kMask = bits::lowMask(3) = 0b111 (0x07) kMask2 = 0x07 | (0x07 << 8) = 0x0707 kMask4 = 0x0707 | (0x0707 << 16) = 0x07070707 kDepMask8 = 0x07070707 | (0x07070707 << 32) = 0x0707070707070707 假设我们有8个3位值：[5, 3, 7, 2, 6, 1, 4, 0]（十进制）二进制：101 011 111 010 110 001 100 000 在64位word中排列（假设从低位开始）： word = 0b...000'100'001'110'010'111'011'101 = 0x0000000000B7B180 (十六进制) = 00000000 00000000 00000000 00000000 00000000 10110111 10110001 10000000 pdep指令会进行如下计算： ●掩码 0x0707070707070707 有24个1位（8个字节×3位） ●PDEP指令将word的低24位（因为掩码有24个1）依次放置到掩码中为1的位置最终输出： eightBytes = 0x0004050607020305 = [0x05, 0x03, 0x07, 0x02, 0x06, 0x01, 0x04, 0x00] (每个字节低3位) 当得到eightBytes后，我们使用： ●_mm_loadl_epi64将eightBytes加载到avx2寄存器的低64位； ●使用_mm256_cvtepu8_epi32将8个8位无符号整数零扩展为8个32位整数，存到256位寄存器eightInts，此时的eightInt并非是数组，而是一个256位数据； ●进入store8Ints(eightInts, i, result); 将eightInts数据切分成的8个32位整数，提取到result数组中。如此，便完成了一轮循环里的数据解包！ 2.2.2.行不连续：使用聚集（gather）操作，代码如下： template int32_t decode1To24( const uint64_t* bits, int32_t bitOffset, const int* rows, int32_t numRows, T* result) { constexpr uint64_t kMask = bits::lowMask(width); int32_t i = 0; const auto masks = as8x32(_mm256_set1_epi32(kMask)); for (; i + 8 <= numRows; i += 8) { auto row = rows[i]; auto endRow = rows[i + 7]; __m256i eightInts; if (width <= 16 && endRow - row == 7) {...} //当8行连续且位宽≤16 } else { eightInts = gather8Sparse (bits, bitOffset, rows, i, masks, result); //非连续或位宽>16 } store8Ints(eightInts, i, result); } } template FOLLY_ALWAYS_INLINE __m256i gather8Sparse( const uint64_t* bits, int32_t bitOffset, const int32_t* rows, int32_t i, __m256si masks, T* result) { constexpr __m256si kMultipliers = {256, 128, 64, 32, 16, 8, 4, 2}; constexpr __m256si kWidthSplat = { width, width, width, width, width, width, width, width}; auto indices = *reinterpret_cast (rows + i) * kWidthSplat + bitOffset; __m256si multipliers; if (width % 8 != 0) { multipliers = (__m256si)_mm256_permutevar8x32_epi32( as256i(kMultipliers), as256i(indices & 7)); } auto byteIndices = indices >> 3; auto data = as8x32(_mm256_i32gather_epi32( reinterpret_cast (bits), as256i(byteIndices), 1)); if (width % 8 != 0) { data = (data * multipliers) >> 8; } return as256i(data & masks); } indices：每个值的位(bit)位置示例：width=3, rows[i..i+7] = [0,2,4,6,8,10,12,14], bitOffset=5 indices = [0,2,4,6,8,10,12,14] × 3 + 5 = [5, 11, 17, 23, 29, 35, 41, 47] multipliers: 当width % 8 != 0, 当位宽width不是8的倍数时，需要读取的位可能不从字节边界开始；由于最后需要data & masks做掩码提取，因此一定要让数据对齐字节边界。以data中的某一个元素举例： data1：.....10110010 indice & 7 (偏移量)：5 则data1在multipliers中对应的数据为： {256, 128, 64, 32, 16, 8, 4, 2} -------------------------------------------------------------------↑ 即 8；由于data1偏移量是5，则它应该从第五位（下标从0开始）开始往后取.....10110010 ----------------------------------------------------------------------------------↑ 如果直接将data1 & masks，由于数据没有从第0位开始，会取得错误的结果；因此需要使用(data * multipliers) >> 8，先让data1左移3位，再右移8位，此时data1的数据就会变成.......101|10010(另外的字节) 这时候就可以正确使用掩码提取。 data：通过byteIndices找到每个数据所在的字节边界，如果非字节对齐，则进一步计算data = (data * multipliers) >> 8;，完成字节对齐。三.优化代码从以上分析可以看出，如果要将sve指令改写，主要有两个重点： 3.1. 代替_pdep_u64指令 _pdep_u64指令是x86独有的以位为单位进行操作的指令。经过调研后，我们决定使用sve2指令：svbdep_64指令进行代替： svbdep_64： svuint64_t svbdep[_u64](svuint64_t op1, svuint64_t op2)是sve2的指令（说明文档地址），以128bit的sve为例， op1是输入数据，op2是掩码，最终return如下图所示：注意，在128bit机器中，使用svbdep时，只能取最低位的那组uint64数据，原因如下：因此调用时svbdep不能一次处理128或256位数据，只能像pdep一样一次处理64位数据。以1<=bitwidth<8的情况为例，优化代码可以写为： auto bit = row * width + bitOffset; auto byte = bit >> 3; auto shift = bit & 7; uint64_t word = *addBytes(bits, byte) >> shift; svuint64_t word_vec = svdup_u64(word); svuint64_t result_vec = svbdep_u64(word_vec, kBdepMask8); eightBytes = svlastb_u64(pg1, result_vec); 3.2. 完成sve版本的gather8Sparse改写：其实思路和avx2的实现一样，只不过将对应指令换成sve的，代码如下： // gather8Sparse_sve: 使用SVE从位压缩流中提取8个稀疏值 // 输入: bits (uint64_t*), bitOffset (int32_t), rows (int32_t*), i (int32_t) // 输出: eightInts (uint32_t[8]) function gather8Sparse_sve(bits, bitOffset, rows, i, eightInts): // 为有效行创建谓词（处理尾部） pg = predicate_from_range(i, i+8) // 32位元素谓词 pg8 = full_predicate(8) // 8个通道始终为真 // 加载8个行号（有符号扩展为零） row_vec = load_32bit_signed(rows + i, pg) // 计算绝对位位置: indices = row_vec * width + bitOffset indices = row_vec * width + bitOffset // 在pg8上饱和/谓词化 // 将位位置转换为字节索引（>> 3） byte_indices = indices >> 3 // 从bits数组按计算的字节索引收集32位字 // 注意: bits被转换为uint32_t*用于收集 data = gather_32bit(bits, byte_indices, pg8) // 如果宽度不是8的倍数，将加载的字移位使值对齐到字节边界 if width % 8 != 0: // 每个可能位偏移（0..7）的预定义乘数 multipliers_table = [256, 128, 64, 32, 16, 8, 4, 2] // 计算字节内偏移量（indices的低3位） bit_in_byte = indices & 7 // 查表获取每个通道的乘数 multipliers = table_lookup(multipliers_table, bit_in_byte) // 先乘后右移8位 -> 提取精确的值位 data = (data * multipliers) >> 8 // 计算低'width'位的掩码 if width == 0: mask_val = 0 else: mask_val = (1 << width) - 1 // 应用掩码保留所需位 result_vec = data & mask_val // 将结果存储到输出数组eightInts[0..7] store_32bit(eightInts, result_vec, pg8) 四.效果因为unpack函数是readWithVisitor的fastPath中的一个环节，当时整个fastPath在arm上还没打通，因此性能验收是临时使用了velox的ut来做校验，提升如下：返回类型T为int32时，平均提升50%+；返回类型T为int64时，平均提升50%+。