ch06Pages

What you build: A page walker for dictionary pages and data pages v1 and v2, with row boundaries and checksums.

The question#

What is inside a column chunk, and how does a reader walk it?

ch04 and ch05 decoded the values in a page, and every column chunk they read held one data page, or a dictionary page and one data page. Real column chunks hold many pages. A writer ends a page when it reaches a size limit, and each page is the smallest unit a reader must decode, and later decompress, to reach any value inside it. Here you walk column chunks page by page, and compare the two versions of the data page.

The experiment#

Walk the pages yourself#

pages.parquet and pages-v2.parquet hold the same sixty orders. The writer was told to keep pages small, so each column chunk holds several. Every seventh country is null. The first file uses data page version 1; the second uses version 2 and puts a checksum in every page header. Appendix B says how pyarrow wrote both.

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.

A column chunk, header by header. The footer says where country’s column chunk starts and how many bytes it takes, headers included. Walk it with the reader’s read_page. It decodes one page header with the Thrift decoder from ch03, then takes as many bytes of body as the header says:

# Walk a column chunk page by page: each header says how long its body is.
import sys
sys.path.insert(0, "python")  # the book's reader, from the repository's python/ directory
from parquet_lab import metadata, pages
from parquet_lab.bytes import ByteReader

data = open("fixtures/pages.parquet", "rb").read()  # try pages-v2.parquet
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]  # country; try 2, which is amount_cents
start = chunk.dictionary_page_offset or chunk.data_page_offset  # the first page's header
end = start + chunk.total_compressed_size  # every page, headers included
print(f"{chunk.dotted_path()}: bytes {start} to {end}")

r = ByteReader(data[start:end], start)
while not r.is_at_end():
    page = pages.read_page(r)  # a Thrift header, then as many bytes of body as it says
    header = page.header_span.end - page.header_span.start
    line = f"{page.page_type} at {page.header_span.start}: {header}-byte header, "
    line += f"{page.compressed_page_size}-byte body, {page.num_values} values"
    if page.v2:  # a version 2 header also counts the page's rows and nulls
        line += f", {page.v2.num_rows} rows, {page.v2.num_nulls} nulls"
    print(line)

//! Walk a column chunk page by page: each header says how long its body is.

use parquet_lab::bytes::ByteReader;
use parquet_lab::metadata::decode_file_metadata;
use parquet_lab::pages;

fn main() {
    let path = "fixtures/pages.parquet"; // try pages-v2.parquet
    let data = std::fs::read(path).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]; // country; try 2, which is amount_cents

    // The first page's header, and the end of every page, headers included.
    let start = match chunk.dictionary_page_offset {
        Some(offset) => offset as usize,
        None => chunk.data_page_offset as usize,
    };
    let end = start + chunk.total_compressed_size as usize;
    println!("{}: bytes {start} to {end}", chunk.dotted_path());
    let mut r = ByteReader::new(&data[start..end], start as u64);
    while !r.is_at_end() {
        // A Thrift header, then as many bytes of body as it says.
        let page = pages::read_page(&mut r).expect("a page");
        let (kind, at) = (&page.page_type, page.header_span.start);
        let header = page.header_span.end - at;
        let (body, values) = (page.compressed_page_size, page.num_values.unwrap_or(0));
        print!("{kind} at {at}: {header}-byte header, {body}-byte body, {values} values");
        if let Some(v2) = page.v2 {
            // A version 2 header also counts the page's rows and nulls.
            print!(", {} rows, {} nulls", v2.num_rows, v2.num_nulls);
        }
        println!();
    }
}

The first page is a dictionary page, and the data pages follow it. Each header starts where the body before it ended, and the last body ends where the footer said the column chunk ends. The loop needs no index and no count of pages, because each header’s body size says where the next header is.

Change pages.parquet to pages-v2.parquet and run the step again. The same rows sit in the same number of pages, but each data page is now a DATA_PAGE_V2, its body is smaller, and its header counts the page’s rows and nulls. Then change the column to 2, which is amount_cents. It has no dictionary page, and its headers are longer. Each carries the page’s minimum and maximum, which for this column are eight-byte integers, where country’s are two-letter strings.

A damaged page. Every page header in pages-v2.parquet carries a CRC-32 of the page’s body. Recompute it for the last page of country, then flip one bit of the body and recompute it:

# Check a page's checksum, then flip one bit of its body and check again.
import sys, zlib
sys.path.insert(0, "python")  # the book's reader, from the repository's python/ directory
from parquet_lab import metadata, pages

data = bytearray(open("fixtures/pages-v2.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
page = pages.walk_pages(data[chunk.start:chunk.end], chunk.start)[-1]  # its last page
start, end = page.body_span.start, page.body_span.end
stored = page.crc & 0xFFFF_FFFF  # the header stores it as a signed 32-bit integer
print(f"{page.page_type} at {page.header_span.start}: its header says {stored:08x}")

def check(label):
    crc = zlib.crc32(data[start:end])  # CRC-32 of the body, as zlib computes it
    print(f"{label}: {crc:08x}, matches the header: {crc == stored}")

check("the body as written")
data[end - 2] ^= 1  # one bit of the values; try another byte, or another bit
check("one bit flipped")
walked = pages.walk_pages(data[chunk.start:chunk.end], chunk.start)
print("the reader's walk:", [p.crc_ok for p in walked])

//! Check a page's checksum, then flip one bit of its body and check again.

use parquet_lab::bytes::crc32;
use parquet_lab::metadata::decode_file_metadata;
use parquet_lab::pages;

fn main() {
    let path = "fixtures/pages-v2.parquet";
    let mut data = std::fs::read(path).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 (from, to) = (chunk.start as usize, chunk.end as usize);
    let walked = pages::walk_pages(&data[from..to], chunk.start).expect("pages");
    let page = walked.last().expect("a page"); // its last page
    let (start, end) = (page.body_span.start as usize, page.body_span.end as usize);
    // The header stores it as a signed 32-bit integer.
    let stored = page.crc.expect("a checksum") as i32 as u32;
    let (kind, at) = (&page.page_type, page.header_span.start);
    println!("{kind} at {at}: its header says {stored:08x}");

    let check = |label: &str, body: &[u8]| {
        let crc = crc32(body); // the book's reader's CRC-32
        println!("{label}: {crc:08x}, matches the header: {}", crc == stored);
    };
    check("the body as written", &data[start..end]);
    data[end - 2] ^= 1; // one bit of the values; try another byte, or another bit
    check("one bit flipped", &data[start..end]);
    let walked = pages::walk_pages(&data[from..to], chunk.start).expect("pages");
    let found: Vec<_> = walked.iter().map(|p| p.crc_ok).collect();
    println!("the reader's walk: {found:?}");
}

The body as written gives the header’s checksum. One flipped bit gives a different one, so the page no longer matches its header. The reader checks every page it walks, and its walk agrees: only the damaged page fails. Python computed the checksum with zlib and Rust with the book’s reader. They agree because CRC-32 is one fixed algorithm.

Change the byte the step flips, or flip two bits, and look at the checksum each time. Then change [-1] to [1] in Python, or last() to get(1) in Rust, to damage the first data page instead, and see which page the reader’s walk marks.

What a page header says#

Every page starts with a Thrift PageHeader. This is the first data page of amount_cents, the column the first step walked when you changed it, read field by field:

compressed_page_size is the field a walker needs: the body’s length, and so where the next header starts. Here the two sizes are equal because the file is not compressed; ch07 makes them differ.

The header also carries the page’s own minimum, maximum and null count. They describe a few rows rather than a whole row group, which makes them more selective, and ch09 uses them to skip pages.

Data page version 1#

These are the pages of country in version 1, with the rows each holds:

The dictionary page is first, and there is only one: every data page’s indices point into it. Each data page body holds three parts, in order:

1. the repetition levels, with a four-byte length in front, if the column can repeat;
2. the definition levels, with a four-byte length in front, if the column can be null;
3. the values, in the page’s encoding.

A version 1 header counts the page’s values, not its rows, so the reader counted the rows by decoding each page’s levels. Under version 1, a page’s whole body is compressed as one block. To count the nulls or the rows in a compressed page, a reader must decompress all of it first.

Data page version 2#

Version 2 changes where things are, not what they mean. The same column chunk:

The first step showed the differences when you ran it on this file, and the two tables set them side by side:

• The level lengths moved into the header. The levels have no four-byte prefix in the body, which is why every body is smaller than its version 1 counterpart.
• The levels are never compressed. Only the values section may be, and the header says whether it is. A reader can count nulls or find row boundaries without decompressing anything.
• The header counts rows and nulls. A reader skipping to a row knows which page holds it from the headers alone.
• A page always starts a new row. In version 1 a row of a repeated column can begin in one page and continue in the next; version 2 forbids that, so pages can be skipped cleanly.

Writers adopted version 2 slowly, and many still write version 1 by default for compatibility. Every reader a file will meet must support version 2 before a writer should produce it.

Checksums#

A page header can carry a CRC-32 of the page body. When it does, a reader can tell a damaged page from one whose values happen to be unusual. Whether a writer includes it depends on the implementation. pyarrow leaves it out unless asked, and the Rust parquet crate never writes it; parquet-java writes it by default. pages-v2.parquet was written by pyarrow with it on.

A checksum helps only a reader that checks it, and checking costs a pass over every body. The bit the second step flipped turns one country into another, and the damaged index still points into the dictionary. A reader that skips the check reads the wrong country without complaint, as the library step at the end of Building it shows.

How big a page should be#

A writer ends a page when its encoded size passes a target, commonly around a megabyte, or when a row count limit is reached. The fixtures use a tiny target so that pages are visible. The trade-off is the same at any size:

• Smaller pages let a reader skip more precisely and decode less to reach one value, but cost more headers and compress slightly worse.
• Larger pages compress better and cost fewer headers, but make every read of one value decode more.

Building it#

The steps looped the reader’s read_page over a column chunk and compared a body with its checksum. This section builds the reader’s parts they used: the walk, the body and its checksum, the version 2 level lengths, and CRC-32 itself. The tabs switch every excerpt on the page between the two languages.

Walking the pages#

The walker reads a page, then the next, until the column chunk ends. It is the first step’s loop:

def walk_pages(chunk: bytes, base: int) -> list[Page]:
    """Walk the pages of a column chunk whose bytes are ``chunk``, starting at file offset
    ``base``.

    Stops at the end of the chunk. A page whose body would run past the end is an error:
    the chunk's size in the footer and the sizes in its page headers must agree.
    """
    r = ByteReader(chunk, base)
    pages = []
    while not r.is_at_end():
        pages.append(read_page(r))
    return pages

pub fn walk_pages(chunk: &[u8], base: u64) -> Result<Vec<Page>, MetadataError> {
    let mut r = ByteReader::new(chunk, base);
    let mut pages = Vec::new();
    while !r.is_at_end() {
        pages.push(read_page(&mut r)?);
    }
    Ok(pages)
}

read_page decodes the header, finds the sub-header that matches the page type, and then takes the body. A body that would run past the end of the chunk is an error: the chunk’s size in the footer and the sizes in its headers must agree. Then it compares the body with the header’s checksum, when there is one, as the second step did:

    # A negative size cannot be read, and too large a size reports where it failed.
    body, body_span = r.read_bytes(compressed if compressed >= 0 else 2**64 - 1)
    crc = opt_int(h, 4)
    # The header stores the CRC as a signed 32-bit integer.
    crc_ok = (crc & 0xFFFF_FFFF) == crc32(body) if crc is not None else None

    let (body, body_span) = r
        .read_bytes(usize::try_from(compressed).unwrap_or(usize::MAX))
        .map_err(|e| MetadataError::Thrift(e.into()))?;
    let crc = opt_int(h, 4);
    // The header stores the CRC as a signed 32-bit integer.
    let crc_ok = crc.map(|c| c as i32 as u32 == crc32(body));

Version 2 bodies#

The only change the column reader needs for version 2 is where the level lengths come from:

        # Version 2 moves the level lengths into the header, and never compresses the
        # levels.
        v2 = page.v2
        rep_len = max(v2.repetition_levels_byte_length, 0) if v2 else None
        def_len = max(v2.definition_levels_byte_length, 0) if v2 else None

        // Version 2 moves the level lengths into the header, and never compresses the
        // levels.
        let (rep_len, def_len) = match page.v2 {
            Some(v2) => (
                Some(v2.repetition_levels_byte_length.max(0) as usize),
                Some(v2.definition_levels_byte_length.max(0) as usize),
            ),
            None => (None, None),
        };

The checksum#

CRC-32, computed a bit at a time, which is slow and short enough to read in full. It gives the same answer as zlib.crc32, which the Python step used:

def crc32(data: bytes) -> int:
    """The CRC-32 checksum of ``data``: the IEEE polynomial, as zlib and Ethernet compute
    it.

    Parquet page headers may carry one over the page body, so a reader can tell a damaged
    page from a page whose values happen to look odd. Computed a bit at a time: slower than
    a table, and short enough to read.
    """
    crc = 0xFFFF_FFFF
    for b in data:
        crc ^= b
        for _ in range(8):
            crc = (crc >> 1) ^ (0xEDB8_8320 if crc & 1 else 0)
    return crc ^ 0xFFFF_FFFF

pub fn crc32(bytes: &[u8]) -> u32 {
    let mut crc = 0xffff_ffffu32;
    for &b in bytes {
        crc ^= u32::from(b);
        for _ in 0..8 {
            let mask = (crc & 1).wrapping_neg();
            crc = (crc >> 1) ^ (0xedb8_8320 & mask);
        }
    }
    !crc
}

Checking it#

Both page fixtures decode to pyarrow’s rows, and every stored checksum verifies and fails after a flipped bit:

python3 -m pytest python/tests

cargo test -p parquet-lab --test fixtures

Ask a library#

pyarrow has no API for pages. Its metadata gives a column chunk’s offsets and size, and it checks every page’s checksum while it reads when you ask it to. The parquet crate reads a column chunk page by page. It checks each checksum when it is built with its crc feature, as the book’s copy is. Both read the file with the bit the second step flipped:

# The same facts from pyarrow: where a column chunk's pages are, and damage refused.
import io
import pyarrow.parquet as pq

path = "fixtures/pages-v2.parquet"
chunk = pq.read_metadata(path).row_group(0).column(1)  # country; pyarrow has no page API
dictionary, first = chunk.dictionary_page_offset, chunk.data_page_offset
start = dictionary or first
print(f"{chunk.path_in_schema}: bytes {start} to {start + chunk.total_compressed_size}")
print(f"the dictionary page at {dictionary}, the first data page at {first}")

def country(source, **options):  # one thread: a refused read must not leave workers behind
    table = pq.read_table(source, columns=["country"], use_threads=False, **options)
    return table["country"].to_pylist()

data = bytearray(open(path, "rb").read())
data[start + chunk.total_compressed_size - 2] ^= 1  # the bit damage_a_page flipped
before, after = country(path), country(io.BytesIO(data))
for row in [row for row, (a, b) in enumerate(zip(before, after)) if a != b]:
    print(f"unchecked, row {row} reads {after[row]!r}, not {before[row]!r}")
try:
    country(io.BytesIO(data), page_checksum_verification=True)
except OSError as e:
    print("checked, it refuses:", e)

//! The same facts from the `parquet` crate: a column chunk's pages, one at a time, and a
//! damaged page refused. The crate checks checksums when built with its `crc` feature.

use bytes::Bytes;
use parquet::column::page::Page;
use parquet::file::reader::{FileReader, SerializedFileReader};

fn open(data: &[u8]) -> SerializedFileReader<Bytes> {
    SerializedFileReader::new(Bytes::copy_from_slice(data)).expect("a Parquet file")
}

fn print_pages(reader: &SerializedFileReader<Bytes>) {
    let row_group = reader.get_row_group(0).expect("a row group");
    for page in row_group.get_column_page_reader(1).expect("pages") {
        match page {
            Ok(Page::DataPageV2 {
                num_rows: rows,
                num_nulls: nulls,
                def_levels_byte_len: def,
                ..
            }) => {
                println!("DATA_PAGE_V2: {rows} rows, {nulls} nulls, {def} level bytes")
            }
            Ok(page) => println!("{}: {} values", page.page_type(), page.num_values()),
            Err(e) => println!("refused: {e}"),
        }
    }
}

fn main() {
    let path = "fixtures/pages-v2.parquet";
    let mut data = std::fs::read(path).expect("run this from the repository's root");
    let reader = open(&data);
    let country = reader.metadata().row_group(0).column(1);
    let (name, (start, size)) = (country.column_path().string(), country.byte_range());
    let end = start + size;
    println!("{name}: bytes {start} to {end}");
    print_pages(&reader);
    data[end as usize - 2] ^= 1; // the bit damage_a_page flipped
    println!("one bit flipped:");
    print_pages(&open(&data));
}

pyarrow finds the column chunk where the first step did. Read without the check, the damaged file gives one row a different country, and nothing says so. With page_checksum_verification, pyarrow refuses the file and names the page. The crate prints each version 2 header’s rows, nulls and level bytes, which agree with the version 2 table, and refuses the damaged page when it reaches it.

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 pages_with_a_library

What this cannot tell you#

What compressed pages look like. Every page here is uncompressed, so a header’s two sizes are equal. ch07 compresses them.

How a reader finds a page without walking. Walking is linear: to reach the last page, a reader reads every header before it. The page index, stored outside the column chunk, lists every page’s offset and row range so a reader can go straight to one. ch09 reads it.

What page size suits your data. The fixtures’ pages are tiny to be visible. The right size for real data depends on how it compresses and how selectively it is read, which ch11 measures.

Index pages. Besides the two data page versions and the dictionary page, the format defines an INDEX_PAGE type that no writer produces. The reader reports one as unexpected rather than guessing at it.

Key takeaways#

Problems#

Three, in exercises/python/pages.py, or in Rust in exercises/src/pages.rs. The first two have tests. The third 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.

6.1 Walk a column chunk. Return the offset of every page, using the book’s Thrift decoder for the headers. The test walks every column chunk of every fixture.

6.2 Check a checksum. Write CRC-32 yourself, and check the fixture’s page checksums with it. The test also damages the pages and expects your check to fail.

6.3 Your own pages. No test: the files are yours. Run PYTHONPATH=python python3 -m parquet_lab pages FILE COLUMN or cargo run -p pqlab -- pages FILE COLUMN on a column of a file your systems write. Record how many pages its first column chunk holds, their sizes, and whether they are version 1 or 2. Then find the writer’s page-size setting in its configuration or documentation, and compare. A good answer explains any gap between the setting and what the pages show; a page far smaller than the target usually means a row count limit was reached first, and the answer should say which limit.

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

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

cargo test -p exercises --test pages -- --ignored

Where to go next#

• Page headers are defined in parquet.thrift, and data page version 2 is described in the format repository’s README.
• ch07 compresses page bodies, and ch09 skips pages without walking to them.

Built from commit c32f48e.