浅谈 Apache ORC 之 Decimal 存储

Apache ORC 官方文档用一句话轻描淡写了 Decimal 的存储，但这句话让我这个 CRUD Boy 看的好苦。

Decimal was introduced in Hive 0.11 with infinite precision (the total number of digits). In Hive 0.13, the definition was change to limit the precision to a maximum of 38 digits, which conveniently uses 127 bits plus a sign bit. The current encoding of decimal columns stores the integer representation of the value as an unbounded length zigzag encoded base 128 varint. The scale is stored in the SECONDARY stream as a signed integer.

要看懂这句话，需要搞明白几个概念：

• 什么是 precision，什么是 scale？
• 什么是 int128？
• Zigzag 编码是什么？

搞懂上面概念后，再来理一理 ORC 存储 Decimal 的逻辑。

Precision 和 Scale 是什么

Precision 表示总共数字的个数，scale 表示 . 右边数字的个数。

例如 Decimal(5, 2)，其中 5 是 precision，2 是 scale。

以 123.45 为例，precision 是 5，scale 是 2。

Int128

int128在 ORC 源码里面的注释是：

  /**
   * Represents a signed 128-bit integer in two's complement.
   * Calculations wrap around and overflow is ignored.
   *
   * For a discussion of the algorithms, look at Knuth's volume 2,
   * Semi-numerical Algorithms section 4.3.1.
   *
   */
  class Int128 {
   private:
     int64_t highbits;
     uint64_t lowbits;
  }

Int128 由 int64_t 和 uint64_t 组成，存储的都是补码（two’s complement）。ORC 为 int128 实现了常规的运算逻辑（加减乘除位移等），具体算法实现见源码 c++/include/orc/Int128.hh。

Zigzag

介绍 Zigzag 编码之前，先复习一下原码、反码和补码。

原码、反码、补码

原码

第一个位表示符号位（0为非负数，1位负数），剩下的位表示值。比如：

$$\begin{aligned}+8=[00001000]_{原}\\-8=[10001000]_{原}\end{aligned}$$

反码（one’s complement）

仍用第一位表示符号（ 0 为非负数，1 为负数），剩下的位，非负数保持不变，负数按位求反。比如：

$$\begin{aligned}+8=[00001000]_{原}=[00001000]_{反}\\-8=[10001000]_{原}=[11110111]_{反}\end{aligned}$$

补码（two’s complement）

第一位依旧是符号位（ 0 为非负数，1 为负数），剩下的位非负数保持不变，负数按位求反并在末位加一。

$$\begin{aligned}+8=[00001000]_{原}=[00001000]_{反}=[00001000]_{补} \\ -8=[10001000]_{原}=[11110111]_{反}=[11111000]_{补}\end{aligned}$$

1+(-1)按照补码的方式，模拟计算一下：

$$\begin{equation*} \begin{aligned} 1 + (-1) &= [00000001]_{补}+[11111111]_{补} \\ &= [00000000]_{补} \\ &= 0 \end{aligned}\end{equation*} $$

补码能有效解决 +0，-0 存在两种表达方式的问题。这样计算机运算的时候，只需要按照统一的逢二进一的原则进行运算就行了，无需关心符号位。

可以简单的认为，计算机中存储数据，都是以补码的形式存储。

为什么要有zigzag

在绝大多数情况下，我们使用到的整数，都是比较小的。比如我的银行卡余额、网站浏览量等等。而在系统之间进行通讯的时候，往往又以 int(4 bytes)或long(8 bytes)为基本传输类型。假设传递一个 (int) 1，需要传输 4 个 Bytes：00000000_00000000_00000000_00000001 32 个bits。这里面除了最后一位是有价值的 1，其它都是无价值的 0。

而 zigzag 的主要作用就是对数据进行压缩，比如上面 1这个例子，使用 zigzag 编码后，只会占用 1 个 byte，用 00000001 表示。

Zigzag 算法实现

接下来具体介绍 zigzag 算法的思想。

对于正整数来讲，如果在数据传输的时候，把多余的 0 去掉（或者是尽可能去掉无意义的 0 ），从有意义 1 开始的数据进行传输，那不是就可以做到数据的压缩了。那怎么压缩无意义的 0 呢？

答案也很简单，比如：

$$00000000\_00000000\_00000000\_00000001$$

这个（int）1，如果只发送一 bit 的 1 或者一个字节00000001，是不是就能压缩很多额外的数据呢？

当然，如果这个世界只有正整数，我们可以方便的做到这一点。可惜，它居然还有负数！！！负数长什么样呢？例如 -1 的补码如下：

$$11111111\_11111111\_11111111\_11111111_{补}$$

前面全是 1，你说怎么弄？！

Zigzag 给出了一个很巧的方法：之前讲补码说过，补码的第一位是符号位，它阻碍了我们对于前导 0 的压缩，那么，我们就把这个符号位放到补码的末尾，其它位整体前移一位。

$$\begin{gather}\textbf{1}1111111\_11111111\_11111111\_11111111_{补}\\\downarrow\\11111111\_11111111\_11111111\_1111111\textbf{1}_{符号后移}\end{gather}$$

但是即使这样，也是很难压缩的，因为数字绝对值越小，他所含的前导 1 越多。于是，这个算法就把负数的所有数据位按位求反，符号位还是保持不变，得到了这样的整数：

$$\begin{gather}(-1)_{10进制}\\\downarrow\\\textbf{1}1111111\_11111111\_11111111\_11111111_{补}\\\downarrow\\11111111\_11111111\_11111111\_1111111\textbf{1}_{符号后移}\\\downarrow\\00000000\_00000000\_00000000\_0000000\textbf{1}_{zigzag}\end{gather}$$

非负数1和负数一样，符号位后移，但是其数据位不需要翻转。

$$\begin{gather}(1)_{10进制}\\\downarrow\\\textbf{0}0000000\_00000000\_00000000\_00000001_{补}\\\downarrow\\00000000\_00000000\_00000000\_0000001\textbf{0}_{符号后移}\\\downarrow\\00000000\_00000000\_00000000\_0000001\textbf{0}_{zigzag}\end{gather}$$

这样一弄，非负数和负数都有相应的表示方法了。上面的处理逻辑写成代码就是如下算法：

int32_t int32ToZigzag(int32_t n) {
    return (n << 1) ^ (n >> 31);
}

n << 1 将整个值左移一位，不管正数、0、负数它们的最后一位就变成了 0。

$$\begin{gather}(1)_{10进制}\newline\downarrow\newline00000000\_00000000\_00000000\_00000001_{补}\newline\downarrow\newline00000000\_00000000\_00000000\_00000010_{左移1位}\end{gather}$$

n >> 31 无论正负，将符号位右移 31 位。因为 int32 是有符号的，所以负数高位补 1，正数高位补0。

$$\begin{gather}(1)_{10进制}\newline\downarrow\newline00000000\_00000000\_00000000\_00000001_{补}\newline\downarrow\newline00000000\_00000000\_00000000\_00000000_{右移31位}\end{gather}$$

最后这一步很巧妙，将两者进行按位异或操作，(n << 1) ^ (n >> 31)：

$$\begin{gather}(1)_{10}\\\downarrow\\00000000\_00000000\_00000000\_00000010_{左移1位}\\\oplus\\00000000\_00000000\_00000000\_00000000_{右移31位}\\=\\00000000\_00000000\_00000000\_00000010_{zigzag}\end{gather}$$

接下来再看看 (-1) 的操作方法：

(n << 1)：

$$\begin{gather}(-1)_{10进制}\newline\downarrow\newline11111111\_11111111\_11111111\_11111111_{补}\newline\downarrow\newline11111111\_11111111\_11111111\_11111110_{左移1位}\end{gather}$$

(n >> 31)：

$$\begin{gather}(-1)_{10进制}\newline\downarrow\newline11111111\_11111111\_11111111\_11111111_{补}\newline\downarrow\newline11111111\_11111111\_11111111\_11111111_{右移31位}\end{gather}$$

$$\begin{gather}(-1)_{10}\\\downarrow\\11111111\_11111111\_11111111\_11111110_{左移1位}\\\oplus\\11111111\_11111111\_11111111\_11111111_{右移31位}\\=\\00000000\_00000000\_00000000\_00000001_{zigzag}\end{gather}$$

可以看到，经过这一步巧妙的算法，前导数都是0了，而且符号位的信息也保留了下来。

Zigzag 还原代码如下：

int32_t zigzagToInt32(int32_t zz) {
    return int32_t(uint32_t(zz)>>1) ^ -(zz & 1);
}

类似的，我们还原的代码就反过来写就可以了。

int32_t(uint32_t(zz)>>1)，这里之所以需要先把 zz 强制转换成 uint32_t 是因为我们是使用无符号的右移操作。否则如果第一位数据位是 1 的话，就会补1。

(zz & 1) 则是把符号位取了出来，0 表示正数，1 表示负数。

-(zz & 1) 取负数，如果是 0，那么取负数还是0，如果是1，取负数就是 -1。

下面以复原 -1 的 zigzag 为例。

int32_t(uint32_t(zz)>>1)：

$$\begin{gather}00000000\_00000000\_00000000\_00000001_{zigzag}\\\downarrow\\00000000\_00000000\_00000000\_00000000_{补}\end{gather}$$

(zz & 1)=1：

$$\begin{gather}00000000\_00000000\_00000000\_00000001_{zigzag}\\\downarrow\\00000000\_00000000\_00000000\_00000001_{补}\end{gather}$$

-(zz & 1)=-1：

$$\begin{gather}11111111\_11111111\_11111111\_11111111_{补}\end{gather}$$

int32_t(uint32_t(zz)>>1) ^ -(zz & 1)：

$$\begin{gather}00000000\_00000000\_00000000\_00000000_{补}\\\oplus\\11111111\_11111111\_11111111\_11111111_{补}\\=\\11111111\_11111111\_11111111\_11111111_{补}\\= -1\end{gather}$$

复原成功。

好了，有了算法对数字进行转换以后，我们就得到了有前导 0 的另外一个整数了，不过它还是一个 4 字节的整数。接下来就要考虑怎么用尽可能少的字节数来表示它，并且还能还原。

比如将 1 转换成的 zigzag如下：

$$00000000\_00000000\_00000000\_00000010_{zigzag}$$

下面最好只需要发送 2 bits（10），或者发送 8 bits（00000010）就好了，反正就是把前面无意义的 0 全部省掉。

Zigzag 引入了一个方法，就是每一个字节的最高位（也就是第一位）用于表示后面是否还有有效数字。

以 -1000 为例：

$$\begin{gather}(-1000)_{十进制}\\\downarrow\\00000000\_00000000\_00000111\_11001111_{zigzag}\end{gather}$$

因为每一个byte的最高位要用于表示后面还有没有有效数字，所以实际上一个1个byte只有7个bit是可以用于存放原始的数字，相当于上面的 zigzag 被转换成如下结构：

$$\begin{gather}00000000\_00000000\_00000111\_11001111_{zigzag}\\\downarrow\\ 00000000\_00000000\_X0001111\_X1001111\end{gather}$$

其中 $X$ 表示后面是否仍然有有效数字（从低位往高位处理）：

• 从左往右数，第二个 $X$ 其实是1，因为后面还有有效数字 0001111。
• 从左往右数，第一个$X$其实是0，因为后面已经没有有效数字了，全都是0。

也就是说，在实际存储数据的时候，只需要两个字节就行了，分别是 [11001111, 00001111]。

上面的逻辑写成代码就是：

void compress(int32_t zz, std::vector<unsigned char>& buffer) {
    while(true) {
        if (zz & (~0x7F)) {
            buffer.emplace_back((zz & 0x7f) | 0x80);
            zz = int32_t(uint32_t(zz) >> 7);
        } else {
            buffer.emplace_back((zz & 0x7f));
            break;
        }
    }
}

里面的 0x7F 是 01111111 ，0x80 是 10000000。

注意 zigzag 右移的时候，要先转换成 uint32_t，进行无符号的右移。

具体逻辑就不啰嗦了，核心的思想就是：

• 每一次循环只处理 7 个 bit。
• 先判断除了末尾的 7 位，高位是否仍然存在有效数字。如果存在有效数字，则取出末尾的 7 位，并在其高外上面加个1，表示后面还存在有效的 byte。否则就在高位加个0，表示后面不存在有效数字，并跳出循环。

buffer 里面的 byte 是从 zigzag的低位往高位存储的。

比如：

$$00000000\_00000000\_00000111\_11001111_{zigzag}$$

存在 buffer 里面的样子就是 [11001111, 00001111]，这点到时候在复原的时候需要考虑到。

从 buffer 里面解压出 zigzag 的代码逻辑如下：

int32_t decompress(std::vector<unsigned char>& buffer) {
    int32_t zz = 0;
    int32_t offset = 0;
    for (const auto& b : buffer) {
        if ((b & 0x80) == 0x80) {
            zz |= int32_t(b&0x7F) << offset;
            offset += 7;
        } else {
            zz |= int32_t(b&0x7F) << offset;
            break;
        }
    }
    return zz;
}

因为存在 buffer 里面的 byte 顺序和原来 zigzag 顺序相反，所以这里需要 int32_t(b&0x7F) << offset 提前对 buffer 里面的 byte 进行位移操作。

ORC Decimal

ORC decimal 有 int128 和 int64 两种存储方法，下面就以 int64 为例。

Decimal64ColumnReader 创建后，会从 orc type 里面读取出 decimal 的 precision 和 scale。

基于 DATA_STREAM创建 valueStream，valueStream 就是用于读取每一行数字。

基于 SECONDARY_STREAM 创建 scaleDecoder，用于读取每一行的数字的 scale。

  Decimal64ColumnReader::Decimal64ColumnReader(const Type& type, StripeStreams& stripe)
      : ColumnReader(type, stripe) {
    scale = static_cast<int32_t>(type.getScale());
    precision = static_cast<int32_t>(type.getPrecision());
    valueStream = stripe.getStream(columnId, proto::Stream_Kind_DATA, true);

    std::unique_ptr<SeekableInputStream> stream =
        stripe.getStream(columnId, proto::Stream_Kind_SECONDARY, true);
    scaleDecoder = createRleDecoder(std::move(stream), true, vers, memoryPool, metrics);
  }

Decimal64ColumnReader::next() 每次读取一批数据：

void Decimal64ColumnReader::next(ColumnVectorBatch& rowBatch, uint64_t numValues, char* notNull) {
  ColumnReader::next(rowBatch, numValues, notNull);
  notNull = rowBatch.hasNulls ? rowBatch.notNull.data() : nullptr;
  Decimal64VectorBatch& batch = dynamic_cast<Decimal64VectorBatch&>(rowBatch);
  int64_t* values = batch.values.data();
  // read the next group of scales
  int64_t* scaleBuffer = batch.readScales.data();
  scaleDecoder->next(scaleBuffer, numValues, notNull);
  batch.precision = precision;
  batch.scale = scale;
  if (notNull) {
    for (size_t i = 0; i < numValues; ++i) {
      if (notNull[i]) {
        readInt64(values[i], static_cast<int32_t>(scaleBuffer[i]));
      }
    }
  } else {
    for (size_t i = 0; i < numValues; ++i) {
      readInt64(values[i], static_cast<int32_t>(scaleBuffer[i]));
    }
  }
}

先通过 scaleDecoder->next(scaleBuffer, numValues, notNull); 读取出这一批数据每一行的 scale，然后调用 readInt64() 读取每一行的数字。

  void readInt64(int64_t& value, int32_t currentScale) {
    value = 0;
    size_t offset = 0;
    while (true) {
      readBuffer();
      unsigned char ch = static_cast<unsigned char>(*(buffer++));
      value |= static_cast<uint64_t>(ch & 0x7f) << offset;
      offset += 7;
      if (!(ch & 0x80)) {
        break;
      }
    }
    value = unZigZag(static_cast<uint64_t>(value));
    if (scale > currentScale && static_cast<uint64_t>(scale - currentScale) <= MAX_PRECISION_64) {
      value *= POWERS_OF_TEN[scale - currentScale];
    } else if (scale < currentScale &&
               static_cast<uint64_t>(currentScale - scale) <= MAX_PRECISION_64) {
      value /= POWERS_OF_TEN[currentScale - scale];
    } else if (scale != currentScale) {
      throw ParseError("Decimal scale out of range");
    }
  }

while 循环里面的代码就是上面 zigzag 的 decompress() 的变种写法。

  while (true) {
    readBuffer();
    unsigned char ch = static_cast<unsigned char>(*(buffer++));
    value |= static_cast<uint64_t>(ch & 0x7f) << offset;
    offset += 7;
    if (!(ch & 0x80)) {
      break;
    }
  }

unZigZag() 就是上面 zigzagToInt32() 的逻辑。

后面就是当 scale 不匹配的时候，乘除一下。

if (scale > currentScale && static_cast<uint64_t>(scale - currentScale) <= MAX_PRECISION_64) {
      value *= POWERS_OF_TEN[scale - currentScale];
    } else if (scale < currentScale &&
               static_cast<uint64_t>(currentScale - scale) <= MAX_PRECISION_64) {
      value /= POWERS_OF_TEN[currentScale - scale];
    } else if (scale != currentScale) {
      throw ParseError("Decimal scale out of range");
    }

比如 123.45，其 ORC valueStream 里面存储的是 12345，scaleDecoder 里面存储的是 2。

此时如果 orc type 里面的 scale 是2，那就不用处理，直接读出来就行了。

如果 orc type 里面的 scale 是1，那么valueStream 里面的值需要除以10，变成1234，decimal 值是123.4。

如果 orc type 里面的scale是3，那么valueStream 里面的值需要乘以10，变成123450，decimal 值是 123.450。

参考链接：

ORC DecimalColumn Specification

深入微服务：2. 研究 Protobuf 时发现一个挺好的算法 — ZigZag

反码和补码的数学原理