上一篇 ClickHouse为什么这么快（一）减少数据扫描范围 我们说到了ClickHouse中使用列存储，每个列都单独存储为一个文件，每个文件都是由一个或多个数据块组成，也就是说：每个文件由一个或多个数组组成，这篇博客就说一说如何使用SSE指令优化在数组上的查询。

问题：为了让性能对比更加明显，我们假定有一个表tab02包含一个uint8类型的列col，完成以下查询：

SELECT count(col) FROM tab02 WHERE col > 0

 由于col的数据存储为一个数组，我们可能会写出如下的代码：

size_t total1 = 0;
for (size_t i = 0; i < TotalCnt; i++)
{
  if (pData[i] > 0)
  {
    total1++;
  }
}

每次循环对一个数进行比较，当满足条件时让统计变量自增1，最后输出统计的变量值即可。这样可以完成任务。

但是，有没有更高性能的实现方法: 当然有，那就是可以一次对多个数进行操作的SSE指令集，在这个对比测试中我们用到了：_mm_setzero_si128，_mm_loadu_si128，_mm_cmpgt_epi8，_mm_movemask_epi8，_mm_popcnt_u32 如果需要运行测试程序请确认你的CPU支持 SSE2和SSE4.2 指令集（确认方法请自行百度）。

对应的实现代码如下：

const size_t SIMD_BYTES = 16;
size_t total2 = 0;
const __m128i zero16 = _mm_setzero_si128();
for (size_t i = 0; i < TotalCnt; i += SIMD_BYTES)
{
  int mask = _mm_movemask_epi8(_mm_cmpgt_epi8(
    _mm_loadu_si128(reinterpret_cast<const __m128i*>(pData + i)), zero16));
  total2 += _mm_popcnt_u32(static_cast<uint32_t>(mask));
}

每个函数的详细的功能介绍请自行百度，这里仅介绍执行逻辑：

执行过程：每次执行对pData中的16个数与zero16中对应位置的数进行比较，如果pData中的数大于zero16中对应位置的值则得1，否则得0；接着将每个比较结果转换为一个bit（也就是mask的值）；最后计算mask值中bit为1的数量，也就是这16条数据中满足条件的数量。

在这个方案中：一次能对16个数进行比较，执行循环的次数仅为前一个实现的1/16，在我的测试中，后面的实现耗时大约只有前面实现耗时的1/8。

在ClickHouse中大量使用SSE指令集优化性能，具体的使用场景请参考源码。这里仅仅是实现了一个性能差别最大的示例来展示SSE指令优化的执行过程。

测试程序在Windows下VS2019编译执行，完整代码如下：

#include <iostream>
#include <Windows.h>
#include <emmintrin.h>
#include <nmmintrin.h>

int main(int argc, char* argv[])
{
  const size_t TotalCnt =  1024 * 1024;
  const size_t SIMD_BYTES = 16;

  LARGE_INTEGER bgTest1, edTest1, bgTest2, edTest2;
  uint8_t* pRaw = new uint8_t[TotalCnt + SIMD_BYTES];
  size_t curMod = reinterpret_cast<uintptr_t>(pRaw) & (SIMD_BYTES - 1);
  uint8_t* pData = pRaw + (curMod == 0 ? 0 : SIMD_BYTES - curMod);
  
  for (size_t i = 0; i < TotalCnt; i++)
  {
    pData[i] = rand() % 10;
  }

  QueryPerformanceCounter(&bgTest1);
  size_t total1 = 0;
  for (size_t i = 0; i < TotalCnt; i++)
  {
    if (pData[i] > 0)
    {
      total1++;
    }
  }
  QueryPerformanceCounter(&edTest1);

  QueryPerformanceCounter(&bgTest2);
  size_t total2 = 0;
  const __m128i zero16 = _mm_setzero_si128();
  for (size_t i = 0; i < TotalCnt; i += SIMD_BYTES)
  {
    int mask = _mm_movemask_epi8(_mm_cmpgt_epi8(
_mm_loadu_si128(reinterpret_cast<const __m128i*>(pData + i)), zero16));
    total2 += _mm_popcnt_u32(static_cast<uint32_t>(mask));
  }
  QueryPerformanceCounter(&edTest2);

  std::cout << "Test1 耗时:" << edTest1.QuadPart - bgTest1.QuadPart 
<< ", total1:" << total1 << std::endl;
  std::cout << "Test2 耗时:" << edTest2.QuadPart - bgTest2.QuadPart 
<< ", total2:" << total2 << std::endl;

  system("pause");
  return 0;
}