storage: support run-length encoding

[skyzh]

It is a common case that users are storing nearly the same value in one column. For example, in TPC-H, we have a CHAR(1) column. There must be duplicated data in this column if we have 1000k rows, and therefore we can use running-length encoding to do some compression.

Guide-level Introduction

RLE encoding is a compression scheme instead of a typical encoding. It can be applied on any column. For example, given a int column:

1, 1, 2, 2, 3, 3
-> PrimitiveBlockBuilder(1, 1, 2, 2, 3, 3)

RLE encoding will check if the current value is the same as the previous one, and generate a RLE map:

1, 1, 2, 2, 3, 3
-> PrimitiveBlockBuilder(1, 2, 3)
-> RLE(2, 2, 2)

Implementation-level Introduction

Builder and Iterator

Contrary to primitive iterators, RLE block builder and iterator are simply a wrapper on primitive iterators. It will look like follows:

pub struct RLEBlockBuilder<B: BlockBuilder> {
  block_builder: B,
  rle_count: Vec<u16>,
  previous_value: Option<B::ArrayType::OwnedItem>
}

When finishing a block, RLEBlockBuilder simply append the RLE count data to the encoded block.

Column

L0 rowsets should not use RLE encoding. We can use distinct value statistics to decide whether to use RLE encoding to the compacted new rowsets.

We should modify factories in column correspondingly:

pub(super) enum BlockIteratorImpl<T: PrimitiveFixedWidthEncode> {
    Plain(PlainPrimitiveBlockIterator<T>),
    PlainNullable(PlainPrimitiveNullableBlockIterator<T>),
    /* new */ RLEPlainNullable(RLEBlockIterator<PlainPrimitiveNullableBlockIterator<T>>)
}

And in proto:

  enum BlockType {
    Plain = 0;
    RunLength = 1;
    ZstdCompress = 2;
    PlainNullable = 3;
    PlainFixedChar = 4;
    PlainVarchar = 5;
    RLEPlainNullable = 6; /* new */
  }