ch05Encodings

What you build: Decoders for dictionary pages and indices, the delta encodings and BYTE_STREAM_SPLIT, each with a trace.

The question#

How do values become compact bytes before any compressor sees them?

ch04 read values that were stored PLAIN: each one written out in full, little-endian, back to back. PLAIN is always correct and rarely small. A column of country codes repeats a handful of strings. A column of order ids goes up by one each row. A column of URLs shares most of every value with the value before it. PLAIN stores all of that repetition in full.

An encoding is a type-aware way of writing a column’s values that exploits patterns like these. It happens before compression (ch07), it knows what the values are, and a reader can decode it quickly. This chapter reads the encodings a writer chooses between, from their bytes.

The experiment#

Read the encodings yourself#

encodings.parquet holds forty orders, and each column was written with the encoding that suits its values. dictionary.parquet holds twelve orders written with the default, dictionary encoding. Appendix B says how pyarrow wrote each.

Each step below is a few lines of code. In Python, run them in the page, change them with Edit, and run them again. In Rust, open them in a Codespace, or run one at a desk with cargo run -p walkthroughs --bin and its name. Each step finds the footer as ch02 did, and the reader’s walk_pages finds a column chunk’s pages as it did in ch04. Every column in both files is required, so a data page’s body holds values and no levels.

A run of deltas. ordered_at is a plain INT64 with no logical type: a count of seconds since 1970, with rows about a minute apart. pyarrow wrote it as DELTA_BINARY_PACKED. Read the start of its page by hand, with two small helpers for the varints: ULEB128, as in the run headers of ch04, and zigzag, as in the footer’s Thrift fields in ch03:

# DELTA_BINARY_PACKED by hand: a header of varints, then a block of bit-packed deltas.
import sys
sys.path.insert(0, "python")  # the book's reader, from the repository's python/ directory
from parquet_lab import metadata, pages

data = open("fixtures/encodings.parquet", "rb").read()
length = int.from_bytes(data[-8:-4], "little")
base = len(data) - 8 - length
md = metadata.decode_file_metadata(data[base:-8], base)
column = 1  # ordered_at; try 0, which is order_id
chunk = md.row_groups[0].columns[column].byte_range()
page = pages.walk_pages(data[chunk.start:chunk.end], chunk.start)[0]  # its one page
body = data[page.body_span.start:page.body_span.end]  # values only: no levels

def uleb128(at):  # seven bits a byte, lowest first, until a byte without its top bit
    n = shift = 0
    while body[at] & 0x80:
        n, at, shift = n | (body[at] & 0x7F) << shift, at + 1, shift + 7
    return n | body[at] << shift, at + 1

def zigzag(at):  # 0, 1, 2, 3, 4 stand for 0, -1, 1, -2, 2: small either side of zero
    n, at = uleb128(at)
    return n >> 1 ^ -(n & 1), at

block, at = uleb128(0)
miniblocks, at = uleb128(at)
count, at = uleb128(at)
first, at = zigzag(at)
print(f"{len(body)} bytes; header {body[:at].hex(' ')}")
print(f"blocks of {block}, {miniblocks} miniblocks, {count} values, first {first}")
smallest, at = zigzag(at)
widths = list(body[at:at + miniblocks])
print(f"min delta {smallest}; widths {widths}")
at += miniblocks  # the first miniblock: each delta less the smallest, in widths[0] bits
width = widths[0]
bits = int.from_bytes(body[at:at + width], "little")  # eight deltas take width bytes
deltas = [smallest + (bits >> i * width & (1 << width) - 1) for i in range(8)]
print("first deltas:", deltas)

//! DELTA_BINARY_PACKED by hand: a header of varints, then a block of bit-packed deltas.

use parquet_lab::metadata::decode_file_metadata;
use parquet_lab::pages;

/// Seven bits a byte, lowest first, until a byte without its top bit.
fn uleb128(body: &[u8], at: &mut usize) -> u64 {
    let (mut n, mut shift) = (0, 0);
    loop {
        let b = body[*at];
        *at += 1;
        n |= u64::from(b & 0x7F) << shift;
        shift += 7;
        if b & 0x80 == 0 {
            return n;
        }
    }
}

/// 0, 1, 2, 3, 4 stand for 0, -1, 1, -2, 2: small either side of zero.
fn zigzag(body: &[u8], at: &mut usize) -> i64 {
    let n = uleb128(body, at);
    (n >> 1) as i64 ^ -((n & 1) as i64)
}

fn main() {
    let data =
        std::fs::read("fixtures/encodings.parquet").expect("run this from the repository's root");
    let n = data.len();
    let length = u32::from_le_bytes(data[n - 8..n - 4].try_into().unwrap()) as usize;
    let base = n - 8 - length;
    let md = decode_file_metadata(&data[base..n - 8], base as u64).expect("a footer");
    let column = 1; // ordered_at; try 0, which is order_id
    let chunk = md.row_groups[0].columns[column].byte_range();
    let (start, end) = (chunk.start as usize, chunk.end as usize);
    let page = &pages::walk_pages(&data[start..end], chunk.start).expect("pages")[0];
    // The column is required, so the body holds values and no levels.
    let body = &data[page.body_span.start as usize..page.body_span.end as usize];

    let mut at = 0;
    let block = uleb128(body, &mut at);
    let miniblocks = uleb128(body, &mut at) as usize;
    let count = uleb128(body, &mut at);
    let first = zigzag(body, &mut at);
    println!("{} bytes; header {:02x?}", body.len(), &body[..at]);
    println!("blocks of {block}, {miniblocks} miniblocks, {count} values, first {first}");
    let smallest = zigzag(body, &mut at);
    let widths = &body[at..at + miniblocks];
    println!("min delta {smallest}; widths {widths:?}");
    at += miniblocks; // the first miniblock: each delta less the smallest, in widths[0] bits
    let width = u32::from(widths[0]);
    let bits = body[at..at + width as usize]
        .iter()
        .rev()
        .fold(0u64, |acc, b| acc << 8 | u64::from(*b));
    let deltas: Vec<i64> = (0..8)
        .map(|i| smallest + (bits >> (i * width) & ((1 << width) - 1)) as i64)
        .collect();
    println!("first deltas: {deltas:?}");
}

The page body is far smaller than the values written PLAIN, at eight bytes each. It opens with a header of four varints: how many values a block holds, how many miniblocks a block is split into, how many values the page holds, and the first value. No other value is stored whole: the rest of the page is differences.

Then comes a block. It starts with the smallest difference between neighbouring values, and one byte per miniblock giving a bit width. Each delta is stored as its excess over the smallest, in that many bits. The first deltas the step unpacked are all close to a minute, so their excess over the smallest fits in two bits. The page’s deltas all fit in the first miniblock, and the others hold none: their width bytes are zero.

Change column to 0, which is order_id. Its values go up by one, so every delta equals the smallest. Look at the bit widths, and at how many bytes the page body takes.

A dictionary. country in dictionary.parquet repeats a handful of strings. Its column chunk holds two pages. Read both:

# A dictionary page and the indices into it, by hand.
import sys
sys.path.insert(0, "python")  # the book's reader, from the repository's python/ directory
from parquet_lab import metadata, pages

data = open("fixtures/dictionary.parquet", "rb").read()
length = int.from_bytes(data[-8:-4], "little")
base = len(data) - 8 - length
md = metadata.decode_file_metadata(data[base:-8], base)
chunk = md.row_groups[0].columns[1].byte_range()  # country
dictionary, page = pages.walk_pages(data[chunk.start:chunk.end], chunk.start)

body = data[dictionary.body_span.start:dictionary.body_span.end]
entries, at = [], 0
while at < len(body):  # PLAIN strings: a four-byte length, then the bytes
    n = int.from_bytes(body[at:at + 4], "little")
    entries.append(body[at + 4:at + 4 + n].decode())
    at += 4 + n
print(dictionary.page_type, dictionary.encoding, entries)

body = data[page.body_span.start:page.body_span.end]
width, header = body[0], body[1]  # the indices' bit width, then a run's ULEB128 header
kind = f"{page.page_type} {page.encoding}: {len(body)} bytes"
print(f"{kind}, width {width}, header {header:02x}")
bits = int.from_bytes(body[2:], "little")  # bit-packed: header >> 1 groups of eight
slots = page.num_values  # try 8 * (header >> 1), to see the group's padding
indices = [bits >> i * width & (1 << width) - 1 for i in range(slots)]
print("indices:", indices)
print("values:", [entries[i] for i in indices])

//! A dictionary page and the indices into it, by hand.

use parquet_lab::metadata::decode_file_metadata;
use parquet_lab::pages;

fn main() {
    let data =
        std::fs::read("fixtures/dictionary.parquet").expect("run this from the repository's root");
    let n = data.len();
    let length = u32::from_le_bytes(data[n - 8..n - 4].try_into().unwrap()) as usize;
    let base = n - 8 - length;
    let md = decode_file_metadata(&data[base..n - 8], base as u64).expect("a footer");
    let chunk = md.row_groups[0].columns[1].byte_range(); // country
    let (start, end) = (chunk.start as usize, chunk.end as usize);
    let found = pages::walk_pages(&data[start..end], chunk.start).expect("pages");
    let (dictionary, page) = (&found[0], &found[1]);
    let bytes_of = |p: &pages::Page| &data[p.body_span.start as usize..p.body_span.end as usize];

    let body = bytes_of(dictionary);
    let (mut entries, mut at) = (Vec::new(), 0);
    while at < body.len() {
        // PLAIN strings: a four-byte length, then the bytes.
        let n = u32::from_le_bytes(body[at..at + 4].try_into().unwrap()) as usize;
        entries.push(String::from_utf8_lossy(&body[at + 4..at + 4 + n]));
        at += 4 + n;
    }
    let encoding = dictionary.encoding.as_deref().unwrap_or("?");
    println!("{} {encoding} {entries:?}", dictionary.page_type);

    let body = bytes_of(page);
    // The indices' bit width, then a run's ULEB128 header.
    let (width, header) = (u32::from(body[0]), body[1]);
    let encoding = page.encoding.as_deref().unwrap_or("?");
    let (kind, size) = (&page.page_type, body.len());
    println!("{kind} {encoding}: {size} bytes, width {width}, header {header:02x}");
    // Bit-packed: header >> 1 groups of eight indices, lowest bits first.
    let bits = body[2..]
        .iter()
        .rev()
        .fold(0u64, |acc, b| acc << 8 | u64::from(*b));
    let slots = page.num_values.expect("a data page"); // try 8 * (header >> 1) as i64
    let indices: Vec<u64> = (0..slots as u32)
        .map(|i| bits >> (i * width) & ((1 << width) - 1))
        .collect();
    println!("indices: {indices:?}");
    let values: Vec<_> = indices.iter().map(|&i| &entries[i as usize]).collect();
    println!("values: {values:?}");
}

The first page is a dictionary page: each distinct country once, written PLAIN, in the order the writer first met it. The data page after it holds no strings at all. It holds one byte giving a bit width, then a run of the RLE / bit-packing hybrid from ch04, with the position of each row’s country in the dictionary. Four entries need two bits each, so the whole column’s values fit in the few bytes the step printed.

Change slots to 8 * (header >> 1), the number of slots the run’s groups hold, and look at the indices after the page’s own.

The reader, step by step. The book’s reader decodes every encoding the steps met, and a few more. For each page it keeps a record of every step: a label, the bytes it read, and what it read from them. Ask it for url, which pyarrow wrote as DELTA_BYTE_ARRAY:

# The book's reader decodes a column, and records each step: a label, bytes, what it read.
import sys, json
sys.path.insert(0, "python")  # the book's reader, from the repository's python/ directory
from parquet_lab import column, logical, metadata, schema

data = open("fixtures/encodings.parquet", "rb").read()  # or dictionary.parquet
length = int.from_bytes(data[-8:-4], "little")
base = len(data) - 8 - length
md = metadata.decode_file_metadata(data[base:-8], base)
name = "url"  # try "order_id", "sku" or "weight_kg"; or "country" in dictionary.parquet
leaves = schema.leaves(schema.build(md.schema))
leaf = next(leaf for leaf in leaves if leaf.path == [name])
read = column.read_column(data, md.row_groups[0].columns[leaf.column], leaf)

def show(v):
    return json.dumps(logical.value_json(leaf.physical_type, leaf.logical_type, v))

if read.dictionary:
    print("dictionary page:", *(show(v) for v, _ in read.dictionary.entries))
for page in read.pages:
    print(page.encoding, "page")
    for step in page.steps:
        raw = data[step.span.start:step.span.end]
        more = f" … ({len(raw)} bytes)" if len(raw) > 8 else ""
        print(f"  {step.label}: {raw[:8].hex(' ')}{more}\n    {step.detail}")
print("values:", *(show(t.value) for t in read.triples[:4]), "…")

//! The book's reader decodes a column, and records each step: a label, bytes, what it read.

use parquet_lab::encoding::hex;
use parquet_lab::logical::value_json;
use parquet_lab::metadata::decode_file_metadata;
use parquet_lab::plain::PlainValue;
use parquet_lab::{column, schema};

fn main() {
    let data = std::fs::read("fixtures/encodings.parquet") // or dictionary.parquet
        .expect("run this from the repository's root");
    let n = data.len();
    let length = u32::from_le_bytes(data[n - 8..n - 4].try_into().unwrap()) as usize;
    let base = n - 8 - length;
    let md = decode_file_metadata(&data[base..n - 8], base as u64).expect("a footer");
    let name = "url"; // try "order_id", "sku" or "weight_kg"; or "country" in dictionary.parquet
    let leaves = schema::leaves(&schema::build(&md.schema).expect("a schema"));
    let leaf = leaves.iter().find(|l| l.path == [name]).expect("a column");
    let chunk = &md.row_groups[0].columns[leaf.column];
    let read = column::read_column(&data, chunk, leaf).expect("a column");
    let show = |v: &PlainValue| value_json(leaf.physical_type, leaf.logical_type.as_ref(), v);

    if let Some(dictionary) = &read.dictionary {
        let entries = dictionary.entries.iter().map(|(v, _)| show(v).to_json());
        println!("dictionary page: {}", entries.collect::<Vec<_>>().join(" "));
    }
    for page in &read.pages {
        println!("{} page", page.encoding);
        for step in &page.steps {
            let raw = &data[step.span.start as usize..step.span.end as usize];
            let shown = hex(&raw[..raw.len().min(8)]);
            let more = (raw.len() > 8).then(|| format!(" … ({} bytes)", raw.len()));
            let more = more.unwrap_or_default();
            println!("  {}: {shown}{more}\n    {}", step.label, step.detail);
        }
    }
    let values = read.triples.iter().take(4).filter_map(|t| t.value.as_ref());
    let values: Vec<String> = values.map(|v| show(v).to_json()).collect();
    println!("values: {} …", values.join(" "));
}

The page holds two runs of DELTA_BINARY_PACKED integers and then some bytes. The first run is the prefix lengths: how many leading bytes each URL shares with the URL before it. The first URL shares none, and the deltas after it say that most of the others share all but their last character. The second run is the lengths of what is left of each URL, and the bytes are those suffixes, back to back. Only the first URL is stored in full.

Change name to "order_id", "sku" or "weight_kg" and run the step again. order_id is the first step’s other column, read by the reader. sku’s strings are all the same length, and its lengths take no bytes beyond their header and block. weight_kg is a float, split into four streams; look at how few different bytes the last stream holds. Then open dictionary.parquet and ask for "country": the reader finds the same dictionary and indices as the second step.

What each encoding saved#

The same measurement for every column of both files:

Every saving in the table comes from a pattern in the values, and so does the one loss. The dictionary made order_id larger than PLAIN: every value is distinct, so the dictionary holds all of them, and the indices come on top.

pyarrow falls back from a dictionary on its size, not on what it saves. It caps the dictionary’s size, and when a column chunk’s dictionary outgrows the cap it writes the remaining pages with another encoding. order_id’s dictionary is far below the default cap, so pyarrow kept it, and the loss stands. A column of distinct values escapes its dictionary only once there are enough of them to fill the cap. parquet-java also checks what the dictionary saves: after the first page, it drops the dictionary if the dictionary and its indices are not smaller than the plain values would be.

Dictionary encoding#

Dictionary encoding replaces each value with a small integer: its position in a list of the column chunk’s distinct values. The list is written once, as PLAIN, in a dictionary page at the start of the column chunk. The data page then holds the indices, packed with the RLE / bit-packing hybrid, after one byte giving their bit width. That is the layout the second step read.

The run header 05 that the step printed is binary 101: a bit-packed run of two groups of eight. The page has twelve rows, so the run ends with four slots of padding. Dictionary encoding also helps a reader: to find rows where country is UK, it can look UK up once in the dictionary and then compare small integers, and if UK is not in the dictionary it can skip the whole column chunk. Some readers do both. The reader in this book does neither, as ch09 notes.

The data page’s encoding is named RLE_DICTIONARY. Files from format version 1 name it PLAIN_DICTIONARY, which is decoded the same way.

DELTA_BINARY_PACKED#

For integers that are sorted or change slowly, storing the differences between values is far cheaper than storing the values. DELTA_BINARY_PACKED writes a header, then blocks. Each block subtracts its smallest delta from all of its deltas, so every adjusted delta is zero or more, and splits them into miniblocks, each bit-packed at the width its largest adjusted delta needs. The block starts with its smallest delta and one byte per miniblock giving that miniblock’s width. This is ordered_at’s page, as the reader decodes it:

The values are about a minute apart, so the smallest delta is large, but the adjusted deltas are all small and fit in two bits. Deltas that never vary, like order_id’s, adjust to zero and pack into a width of zero: a miniblock with no bytes at all.

The delta encodings for strings#

Two encodings apply the same idea to byte arrays.

DELTA_LENGTH_BYTE_ARRAY writes every value’s length first, as DELTA_BINARY_PACKED, then all the values’ bytes back to back. The sku column’s values are all the same length, so the lengths cost almost nothing, and the bytes are stored without a length in front of each.

DELTA_BYTE_ARRAY writes, for each value, how many leading bytes it shares with the value before it, then only the rest. Sorted strings with common prefixes, such as URLs, paths and hierarchical keys, shrink the most. These are the first of url’s prefix lengths and suffixes, added up from the deltas the third step printed:

A URL that carries into a new digit, such as the step from 109 to 110, shares one byte fewer and stores two.

BYTE_STREAM_SPLIT#

BYTE_STREAM_SPLIT does not make anything smaller. It rearranges a column of fixed-width values into one stream per byte position: every value’s first byte, then every value’s second byte, and so on. Each byte position varies in its own way:

The last stream holds the sign and most of the exponent, and has almost no variety. The stream before it, the top of the mantissa, varies most. The two low streams repeat a few bytes, because these weights go up in steps of a twentieth, and the mantissa of such a fraction repeats one bit pattern. In measured data the low bytes usually vary far more. Written PLAIN, the exponent bytes are scattered one in every four. Written split, they sit together, where a compressor can find the repetition. ch07 measures what that is worth.

Which encodings writers use#

Most writers try dictionary encoding first for every column, and fall back when a dictionary grows too large. Repetition and definition levels always use the RLE / bit-packing hybrid. The delta encodings and BYTE_STREAM_SPLIT are usually opt-in, set per column; pyarrow wrote encodings.parquet only because the generator asked, and only with dictionary encoding turned off. The encodings a column chunk used are recorded in its metadata, so a reader never has to guess.

Building it#

The steps read a delta header and a dictionary by hand, then asked the reader to decode a column. This section builds the reader’s parts they used: the dictionary page and its indices, the delta decoder, and the loop that rebuilds strings from shared prefixes. The tabs switch every excerpt on the page between the two languages.

A dictionary page, then indices#

The column reader decodes a dictionary page, when the column chunk has one, before any data page, and keeps its entries with the bytes each came from:

        if page.page_type == "DICTIONARY_PAGE":
            # The dictionary: every distinct value, PLAIN-encoded, once.
            count = max(page.num_values or 0, 0)
            plain_body = _decompress(codec, body, page, page.uncompressed_page_size)
            try:
                entries = plain.decode(
                    leaf.physical_type, leaf.type_length, plain_body, page.body_span.start, count
                )
            except ValueError as e:
                raise ColumnError(f"dictionary: {e}") from e
            if codec != "UNCOMPRESSED":
                entries = [(v, page.body_span) for v, _ in entries]
            out.dictionary = DictionaryPage(page, entries)
            continue

            "DICTIONARY_PAGE" => {
                // The dictionary: every distinct value, PLAIN-encoded, once.
                let count = page.num_values.unwrap_or(0).max(0) as usize;
                let plain_body = decompress(codec, body, &page, page.uncompressed_page_size)?;
                let mut entries = plain::decode(
                    leaf.physical_type,
                    leaf.type_length,
                    &plain_body,
                    page.body_span.start,
                    count,
                )
                .map_err(|e| ColumnError(format!("dictionary: {e}")))?;
                if codec != "UNCOMPRESSED" {
                    for (_, s) in &mut entries {
                        *s = page.body_span;
                    }
                }
                out.dictionary = Some(DictionaryPage { page, entries });
                continue;
            }

A data page’s indices are the hybrid from ch04, with the bit width in the byte before them. Each decoded value keeps two spans: its dictionary entry, and the run that held its index. The bit width and each run become a step, which the third step prints for country:

        # One byte of bit width, then the indices as an RLE / bit-packing hybrid with no length.
        if not data:
            raise DecodeError(base, "an empty page where the index bit width should be")
        width = data[0]
        runs = rle.decode(data[1:], base + 1, width, count)
        steps = [
            Step(
                "bit width",
                Span(base, base + 1),
                f"{width} bits per index: enough for {len(dictionary)} dictionary entries",
            )
        ]
        found = []
        for n, run in enumerate(runs):
            steps.append(
                Step(
                    f"run {n}",
                    Span(run.header.start, run.body.end),
                    f"{run.kind}: indices {run.values}",
                )
            )
            for i in run.values:
                if i >= len(dictionary):
                    raise DecodeError(
                        run.body.start, f"index {i} is past the dictionary's {len(dictionary)} entries"
                    )
                v, s = dictionary[i]
                found.append(DecodedValue(v, s, [run.body]))

            // One byte of bit width, then the indices as an RLE / bit-packing hybrid with no length.
            let Some(&width) = bytes.first() else {
                return wrong(base, "an empty page where the index bit width should be");
            };
            let runs = rle::decode(&bytes[1..], base + 1, u32::from(width), count).map_err(e)?;
            let mut steps = vec![Step::new(
                "bit width",
                Span::new(base, base + 1),
                format!(
                    "{width} bits per index: enough for {} dictionary entries",
                    dict.len()
                ),
            )];
            let mut values = Vec::with_capacity(count);
            for (n, run) in runs.iter().enumerate() {
                steps.push(Step::new(
                    &format!("run {n}"),
                    Span::new(run.header.start, run.body.end),
                    format!("{}: indices {:?}", run.kind.name(), run.values),
                ));
                for &i in &run.values {
                    let Some((v, s)) = dict.get(i as usize) else {
                        return wrong(
                            run.body.start,
                            &format!("index {i} is past the dictionary's {} entries", dict.len()),
                        );
                    };
                    values.push(DecodedValue {
                        value: v.clone(),
                        span: *s,
                        extra: vec![run.body],
                    });
                }
            }

Deltas#

The delta decoder reads the header’s four varints, as the first step did, then block after block until it has the value count. Each block opens with its smallest delta and the miniblocks’ bit widths:

def binary_packed(data: bytes, base: int) -> Integers:
    """Integers from DELTA_BINARY_PACKED, with the steps that produced them."""
    r = ByteReader(data, base)
    steps = []
    start = r.offset()
    block_size = r.read_uleb128()
    miniblocks = max(r.read_uleb128(), 1)
    total = r.read_uleb128()
    first = r.read_zigzag()
    steps.append(
        Step(
            "header",
            Span(start, r.offset()),
            f"blocks of {block_size} values in {miniblocks} miniblocks; {total} values; the first is {first}",
        )
    )
    per_miniblock = block_size // miniblocks
    out: list[tuple[int, Span]] = []
    if total == 0:
        return Integers(out, steps, r.offset())
    out.append((first, Span(start, r.offset())))
    prev = first
    block = 0
    while len(out) < total:
        block_start = r.offset()
        min_delta = r.read_zigzag()
        widths = list(r.read_bytes(miniblocks)[0])
        steps.append(
            Step(
                f"block {block}",
                Span(block_start, r.offset()),
                f"smallest delta {min_delta}; bit widths {widths}",
            )
        )

pub fn binary_packed(bytes: &[u8], base: u64) -> Result<Integers, BytesError> {
    let mut r = ByteReader::new(bytes, base);
    let mut steps = Vec::new();
    let start = r.offset();
    let block_size = r.read_uleb128()? as usize;
    let miniblocks = r.read_uleb128()?.max(1) as usize;
    let total = r.read_uleb128()? as usize;
    let first = r.read_zigzag()?;
    steps.push(Step::new(
        "header",
        Span::new(start, r.offset()),
        format!(
            "blocks of {block_size} values in {miniblocks} miniblocks; {total} values; the first is {first}"
        ),
    ));
    let per_miniblock = block_size / miniblocks;
    let mut out = Vec::with_capacity(total);
    if total == 0 {
        return Ok(Integers {
            values: out,
            steps,
            end: r.offset(),
        });
    }
    out.push((first, Span::new(start, r.offset())));
    let mut prev = first;
    let mut block = 0;
    while out.len() < total {
        let block_start = r.offset();
        let min_delta = r.read_zigzag()?;
        let (widths, _) = r.read_bytes(miniblocks)?;
        let widths = widths.to_vec();
        steps.push(Step::new(
            &format!("block {block}"),
            Span::new(block_start, r.offset()),
            format!("smallest delta {min_delta}; bit widths {widths:?}"),
        ));

Each miniblock holds a fixed number of deltas at its own width. The decoder unpacks each one’s bits, lowest first, adds the smallest delta back, and adds the result to the previous value. A miniblock past the value count has a width byte and no body:

        for m, w in enumerate(widths):
            if len(out) >= total:
                break  # Unneeded miniblocks have a width byte and no body.
            raw, span = r.read_bytes(per_miniblock * w // 8)
            deltas = []
            for i in range(per_miniblock):
                if len(out) >= total:
                    break
                packed = 0
                for bit in range(w):
                    at = i * w + bit
                    if raw[at // 8] >> (at % 8) & 1:
                        packed |= 1 << (bit % 64)
                delta = _i64(min_delta + _i64(packed))
                prev = _i64(prev + delta)
                deltas.append(delta)
                out.append((prev, span))
            steps.append(Step(f"block {block}, miniblock {m}", span, f"{w} bits per delta; deltas {deltas}"))

        for (m, &w) in widths.iter().enumerate() {
            if out.len() >= total {
                break; // Unneeded miniblocks have a width byte and no body.
            }
            let len = per_miniblock * w as usize / 8;
            let (raw, span) = r.read_bytes(len)?;
            let mut deltas = Vec::new();
            for i in 0..per_miniblock {
                if out.len() >= total {
                    break;
                }
                let mut packed: u64 = 0;
                for bit in 0..w as usize {
                    let at = i * w as usize + bit;
                    if raw[at / 8] >> (at % 8) & 1 == 1 {
                        packed |= 1 << bit;
                    }
                }
                let delta = min_delta.wrapping_add(packed as i64);
                prev = prev.wrapping_add(delta);
                deltas.push(delta);
                out.push((prev, span));
            }
            steps.push(Step::new(
                &format!("block {block}, miniblock {m}"),
                span,
                format!("{w} bits per delta; deltas {deltas:?}"),
            ));
        }

DELTA_BYTE_ARRAY is two decoders and one loop. The prefix lengths are read with binary_packed, the suffixes with the decoder for DELTA_LENGTH_BYTE_ARRAY, and each value is the previous value’s prefix followed by its suffix:

    prev = b""
    values = []
    for (prefix, prefix_span), suffix in zip(prefixes.values, suffixes.values, strict=False):
        value = prev[: min(_size(prefix), len(prev))] + suffix.value
        prev = value
        values.append(DecodedValue(value, suffix.span, [prefix_span, *suffix.extra]))
    return Decoded(values, steps, suffixes.end)

    let mut prev: Vec<u8> = Vec::new();
    let mut values = Vec::with_capacity(prefixes.len());
    for ((prefix, prefix_span), suffix) in prefixes.iter().zip(suffixes.values) {
        let PlainValue::Bytes(tail) = &suffix.value else {
            continue;
        };
        let keep = (*prefix as usize).min(prev.len());
        let mut v = prev[..keep].to_vec();
        v.extend_from_slice(tail);
        prev = v.clone();
        let mut extra = vec![*prefix_span];
        extra.extend(suffix.extra);
        values.push(DecodedValue {
            value: PlainValue::Bytes(v),
            span: suffix.span,
            extra,
        });
    }
    Ok(Decoded {
        values,
        steps,
        end: suffixes.end,
    })
}

Checking it#

Every column of both fixtures is decoded from its bytes, reassembled into records, and compared with the rows pyarrow reported:

python3 -m pytest python/tests -k records_rebuilt

cargo test -p parquet-lab --test fixtures records_rebuilt

Ask a library#

A library decodes every encoding for you, and records in the metadata which ones each column chunk used. pyarrow can also keep a dictionary-encoded column as a dictionary when it reads it. The parquet crate reads a column chunk page by page, and says each page’s type and encoding:

# The same facts from pyarrow: every column chunk's encodings, and a dictionary as read.
import pyarrow.parquet as pq

md = pq.read_metadata("fixtures/encodings.parquet").row_group(0)
for i in range(md.num_columns):
    print(md.column(i).path_in_schema, md.column(i).encodings)
path = "fixtures/dictionary.parquet"
table = pq.read_table(path, columns=["country"], read_dictionary=["country"])
country = table.column("country").chunk(0)  # an Arrow DictionaryArray
print("dictionary:", country.dictionary.to_pylist())
print("indices:", country.indices.to_pylist())

//! The same facts from the `parquet` crate: each column chunk's encodings, and the pages of a
//! dictionary-encoded column, the dictionary page first.

use parquet::file::reader::{FileReader, SerializedFileReader};

fn open(path: &str) -> SerializedFileReader<std::fs::File> {
    let file = std::fs::File::open(path).expect("run this from the repository's root");
    SerializedFileReader::new(file).expect("a Parquet file")
}

fn main() {
    let reader = open("fixtures/encodings.parquet");
    for chunk in reader.metadata().row_group(0).columns() {
        let encodings: Vec<_> = chunk.encodings().collect();
        println!("{} {encodings:?}", chunk.column_path().string());
    }
    let reader = open("fixtures/dictionary.parquet");
    let row_group = reader.get_row_group(0).expect("a row group");
    let country = row_group.metadata().column(1);
    let offset = country.dictionary_page_offset();
    println!(
        "{}: dictionary page at {offset:?}",
        country.column_path().string()
    );
    for page in row_group
        .get_column_page_reader(1)
        .expect("country's pages")
    {
        let page = page.expect("a page");
        println!(
            "{} {} {}",
            page.page_type(),
            page.encoding(),
            page.num_values()
        );
    }
}

Both list each column chunk’s encodings from its metadata. RLE appears beside the encoding of the values because a data page’s header names an encoding for its levels, and pyarrow records it even for a required column that stores none. pyarrow’s read_dictionary hands back an Arrow DictionaryArray, whose dictionary and indices are the ones the second step read by hand. The crate’s metadata gives the offset of the dictionary page, and its page reader returns that page first, then the data page with its count of values.

In Python, the first run loads pyarrow into the page, a much larger download than the other steps. In Rust the step needs the parquet crate, in walkthroughs/libraries, so it runs in a Codespace or at a desk:

cargo run -q --manifest-path walkthroughs/libraries/Cargo.toml --bin encodings_with_a_library

What this cannot tell you#

How much encoding saves on your data. The fixtures’ columns were chosen to suit their encodings. Real columns are messier: a sorted id column with gaps, a URL column in random order. The savings in the table are what these values gave, and yours will differ.

Which encoding a writer should choose. That depends on what the values look like, on what the compressor adds afterwards, and on which readers must open the file. ch07 and ch11 weigh those.

How fast decoding is. This reader decodes one value at a time so that every step is visible. Production readers decode whole miniblocks and runs with vectorised code, and their speed is a property of their implementation, not of anything measured here.

Every encoding and type pairing. The reader decodes the encodings writers use for data pages. It does not decode the deprecated BIT_PACKED level encoding, or RLE for boolean values, which only data page version 2 uses; ch06 returns to that.

Key takeaways#

Problems#

Four, in exercises/python/encodings.py, or in Rust in exercises/src/encodings.rs. The first three have tests. The fourth has none.

The editor below holds the Python stubs and keeps your edits in this browser. Run the tests runs the graders in the page, the same tests that run at a desk.

5.1 DELTA_BINARY_PACKED. Decode it. The test uses the fixture’s integer columns and hundreds of generated encodings with other block sizes, miniblock counts and bit widths.

5.2 Shared prefixes. Rebuild DELTA_BYTE_ARRAY’s values from their prefix lengths and suffixes. The test uses the fixture’s URLs and generated sorted strings.

5.3 Dictionary indices. Read a dictionary-encoded page’s bit width and indices, and return the values. You may use your hybrid decoder from problem 4.1.

5.4 Your own columns. No test: the columns are yours. Pick a Parquet file your systems write and run PYTHONPATH=python python3 -m parquet_lab encodings FILE COLUMN or cargo run -p pqlab -- encodings FILE COLUMN for each column. For each, record the encoding the writer chose and the stored size against PLAIN. Then name one column where a different encoding would suit the values better, and say what pattern in the values makes you think so. A good answer checks the claim by rewriting that column with the other encoding and measuring; if the saving does not appear, say what the values do that you did not expect.

For the Rust stubs, or to work at a desk, run the chapter’s graders with these commands. Add -k problem_5_1 in Python, or problem_5_1 before the -- in Rust, to run one problem:

python3 -m pytest exercises/python/tests/test_encodings.py --problems

cargo test -p exercises --test encodings -- --ignored

Where to go next#

• Every encoding is specified in the format repository’s Encodings.md.
• The delta encoding’s block and miniblock layout follows Lemire and Boytsov’s Decoding billions of integers per second through vectorization.
• ch06 looks at the pages these values live in, and ch07 at what a compressor adds after encoding.

Built from commit c32f48e.