ch09Skipping data

What you build: Conditions judged against statistics, Bloom filters and the page index, with the bytes each saves.

The question#

Given a predicate, which bytes can a reader prove it does not need?

ch08 established when a footer’s minimum and maximum can be trusted. This chapter puts them to work. A query arrives with a condition, such as order_id = 431. Before the reader fetches a column chunk or a page, it asks whether the metadata proves that nothing there can satisfy the condition. If it does, the reader skips it.

One rule governs every decision. A skip must be certain. A reader that skips a page holding a matching row returns a wrong answer, silently. A reader that reads a page it could have skipped only wastes bytes. So every mechanism here answers “skip” or “cannot rule it out”, never “no match” on a guess.

The experiment#

The same orders, sorted and shuffled#

pruning-sorted.parquet and pruning-shuffled.parquet hold the same orders, written the same way: four row groups, pages of forty rows, a page index for every column, and a Bloom filter for customer_id. The first is in order_id order; the second is shuffled. Appendix B says how pyarrow wrote them.

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 decodes it with the reader’s decode_file_metadata from ch03.

Ranges in the footer. The query is order_id = 431. Every row group’s statistics give the smallest and largest order_id it holds. Read them, and ask of each row group whether 431 could be in it:

# Read each row group's range of order_id from the footer: can it hold order 431?
import sys
sys.path.insert(0, "python")  # the book's reader, from the repository's python/ directory
from parquet_lab import metadata

path = "fixtures/pruning-sorted.parquet"  # try pruning-shuffled.parquet
data = open(path, "rb").read()
length = int.from_bytes(data[-8:-4], "little")
base = len(data) - 8 - length
md = metadata.decode_file_metadata(data[base:-8], base)
wanted = 431  # order_id = 431; try 200 and 201

for g, row_group in enumerate(md.row_groups):
    s = row_group.columns[0].statistics  # order_id, an INT64
    low = int.from_bytes(s.min_value, "little", signed=True)  # eight bytes, least first
    high = int.from_bytes(s.max_value, "little", signed=True)
    verdict = "read" if low <= wanted <= high else "skip"
    print(f"row group {g}: order_id {low} to {high}, {verdict}")

//! Read each row group's range of order_id from the footer: can it hold order 431?

use parquet_lab::metadata::decode_file_metadata;

fn main() {
    let path = "fixtures/pruning-sorted.parquet"; // try pruning-shuffled.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 wanted = 431; // order_id = 431; try 200 and 201

    for (g, row_group) in md.row_groups.iter().enumerate() {
        let chunk = &row_group.columns[0]; // order_id, an INT64
        let s = chunk.statistics.as_ref().expect("statistics");
        // Eight bytes each, least significant first.
        let low: [u8; 8] = s.min_value.as_deref().unwrap().try_into().unwrap();
        let high: [u8; 8] = s.max_value.as_deref().unwrap().try_into().unwrap();
        let (low, high) = (i64::from_le_bytes(low), i64::from_le_bytes(high));
        let may_hold = (low..=high).contains(&wanted);
        let verdict = if may_hold { "read" } else { "skip" };
        println!("row group {g}: order_id {low} to {high}, {verdict}");
    }
}

Each row group covers its own band of order numbers, and only row group 2’s band holds 431. The other three are skipped on the footer’s word alone, before a byte of their data is fetched. The test compares 431 with a minimum and a maximum, and nothing else.

Change wanted to 200, then to 201: each lands in one row group, the last of one band or the first of the next. Then change the file to pruning-shuffled.parquet. Every row group now spans nearly every order number, and every one must be read.

When a range rules a condition out#

A column chunk’s statistics, or a page’s, give a minimum, a maximum and a null count. Each comparison is ruled out by a different fact about them:

Null satisfies no comparison. A chunk whose values are all null is skipped for any condition but is null, and it has no minimum or maximum to consult.

Sorted data#

The reader planned SELECT * on both files with each of the conditions below, using every mechanism this chapter describes:

Sorting is what makes statistics work. In the sorted file each row group covers a narrow band of order numbers, and a condition on order_id rules out all but a few. In the shuffled file the same statistics are true and useless: every band is almost the whole range. Columns nobody sorted by, such as customer_id and amount_cents, behave alike in both files. This is why writers are told to sort by the column queries filter on most, and why ch11 returns to it.

The page index#

Row group statistics decide whole row groups. The page index decides pages. It is written after the row groups, in two structures per column chunk: a ColumnIndex with each page’s minimum, maximum and null count, and an OffsetIndex with each page’s position, size and first row. The reader’s page_index module reads both. Ask them about order_id in row group 2, the one row group the first step kept:

# Read the page index of order_id in one row group: each page's range and first row.
import sys
sys.path.insert(0, "python")  # the book's reader, from the repository's python/ directory
from parquet_lab import metadata, page_index

path = "fixtures/pruning-sorted.parquet"  # try pruning-shuffled.parquet
data = open(path, "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[2].columns[0]  # order_id in row group 2
ci = page_index.column_index(data, chunk)  # per page: a minimum, a maximum, a null count
oi = page_index.offset_index(data, chunk)  # per page: an offset, a size, a first row
print(f"{len(oi.pages)} pages, boundary order {ci.boundary_order}")
wanted = 431

for i, page in enumerate(oi.pages):
    low = int.from_bytes(ci.min_values[i], "little", signed=True)
    high = int.from_bytes(ci.max_values[i], "little", signed=True)
    verdict = "read" if low <= wanted <= high else "skip"
    end = page.offset + page.compressed_page_size
    print(f"page {i}: order_id {low} to {high}, {verdict}; "
          f"rows from {page.first_row_index}, bytes {page.offset} to {end}")

//! Read the page index of order_id in one row group: each page's range and first row.

use parquet_lab::metadata::decode_file_metadata;
use parquet_lab::page_index::{column_index, offset_index};

fn main() {
    let path = "fixtures/pruning-sorted.parquet"; // try pruning-shuffled.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");
    // order_id in row group 2. Per page, its ColumnIndex holds a minimum, a maximum and a
    // null count; its OffsetIndex an offset, a size and a first row.
    let chunk = &md.row_groups[2].columns[0];
    let ci = column_index(&data, chunk).unwrap().expect("a ColumnIndex");
    let oi = offset_index(&data, chunk).unwrap().expect("an OffsetIndex");
    let (pages, order) = (oi.pages.len(), &ci.boundary_order);
    println!("{pages} pages, boundary order {order}");
    let wanted = 431;

    for (i, page) in oi.pages.iter().enumerate() {
        let low = i64::from_le_bytes(ci.min_values[i][..].try_into().unwrap());
        let high = i64::from_le_bytes(ci.max_values[i][..].try_into().unwrap());
        let may_hold = (low..=high).contains(&wanted);
        let verdict = if may_hold { "read" } else { "skip" };
        let (start, end) = (page.offset, page.offset + page.compressed_page_size);
        let first = page.first_row_index;
        print!("page {i}: order_id {low} to {high}, {verdict}; ");
        println!("rows from {first}, bytes {start} to {end}");
    }
}

The row group’s pages cover disjoint bands of forty order numbers, and only the first can hold 431. The ColumnIndex also records that its bounds are in ASCENDING order, which lets a reader binary-search them rather than test every page. The OffsetIndex says where each page’s bytes are and which row it starts at, so the reader can fetch that page without the pages after it.

Change the file to pruning-shuffled.parquet. The boundary order is UNORDERED, each page covers nearly every order number, and every page must be read.

A page skipped in the condition’s column means rows skipped, not bytes skipped in every column. The reader turns the kept pages into row ranges using their first rows, then asks each other column’s OffsetIndex which of its pages hold those rows. The fixtures cut every column’s pages at the same row counts, so the pages line up. Writers that cut pages by size give each column different boundaries, and the row ranges are what connects them.

Bloom filters#

Statistics cannot rule out a value inside a chunk’s range. Customer numbers fill most of their range in every row group, so customer_id = 424242 survives every minimum and maximum. A Bloom filter answers a different question: was this exact value ever added?

The filter is a bitset of 32-byte blocks. To add a value, the writer hashes its PLAIN bytes with xxHash64. The hash’s high half picks a block; its low half, multiplied by eight fixed constants, picks one bit in each of the block’s eight 32-bit words; the writer sets those bits. To test a value, the reader computes the same hash and checks the same bits. One clear bit proves the value was never added. All set means it may have been: other values may have set those bits.

Probe each row group’s filter in the shuffled file for customer 424242 with the reader’s bloom module:

# Probe each row group's Bloom filter on customer_id for one customer number.
import struct, sys
sys.path.insert(0, "python")  # the book's reader, from the repository's python/ directory
from parquet_lab import bloom, metadata

path = "fixtures/pruning-shuffled.parquet"  # try pruning-sorted.parquet
data = open(path, "rb").read()
length = int.from_bytes(data[-8:-4], "little")
base = len(data) - 8 - length
md = metadata.decode_file_metadata(data[base:-8], base)
value = struct.pack("<q", 424242)  # an INT64's PLAIN bytes; try another customer
print(f"xxHash64 of {value.hex(' ')}: {bloom.xxh64(value):016x}")

for g, row_group in enumerate(md.row_groups):
    f = bloom.read(data, row_group.columns[1])  # customer_id's filter
    # The hash's high half picks a block, its low half a bit in each of its words.
    p = f.probe(value)
    print(f"row group {g}: block {p.block} of {f.num_blocks()}, bits {p.bits}")
    print(f"  may contain it: {p.may_contain}")

//! Probe each row group's Bloom filter on customer_id for one customer number.

use parquet_lab::bloom::{read, xxh64};
use parquet_lab::metadata::decode_file_metadata;

fn main() {
    let path = "fixtures/pruning-shuffled.parquet"; // try pruning-sorted.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 value = 424242i64.to_le_bytes(); // an INT64's PLAIN bytes; try another customer
    println!("xxHash64 of {value:02x?}: {:016x}", xxh64(&value, 0));

    for (g, row_group) in md.row_groups.iter().enumerate() {
        let chunk = &row_group.columns[1]; // customer_id
        let f = read(&data, chunk).unwrap().expect("a Bloom filter");
        // The hash's high half picks a block, its low half a bit in each of its words.
        let p = f.probe(&value);
        let (block, blocks, bits) = (p.block, f.num_blocks(), p.bits);
        println!("row group {g}: block {block} of {blocks}, bits {bits:?}");
        println!("  may contain it: {}", p.may_contain);
    }
}

Every row group gets the same hash and, since every filter has the same number of blocks, the same block. The bits differ, because each row group’s customers set different ones. In every row group at least one of the eight bits is clear, so 424242 was never added to any of them, and the reader skips the whole file for it.

Change 424242 to 426545, the customer in the shuffled file’s first row: row group 0’s filter has all eight bits set, as it must, since a filter never forgets a value it was given. Then change the file to pruning-sorted.parquet and probe 424242 again. One row group’s filter has all eight bits set, although no order in the file is from that customer. That is a false positive.

The reader measured how often that happens. It probed every filter in the shuffled file with thousands of customer numbers its row group does not hold:

The writer sized each filter for the false-positive rate the table’s note gives, and the measured rate is lower, because the filter’s size is rounded up. False positives cost a read and nothing else: the sorted file’s false positive makes the reader read a row group with no match in it. That is why, in the table under Sorted data, the sorted file reads bytes for customer_id = 424242 and the shuffled one reads none. A filter never produces a false negative, so it never costs a row.

What each mechanism adds#

The reader’s prune.plan puts the three mechanisms together. It takes a condition, the columns to read, and which mechanisms it may use, and decides row group by row group, saying why:

# Plan SELECT * WHERE order_id = 431 with the reader, one mechanism at a time.
import sys
sys.path.insert(0, "python")  # the book's reader, from the repository's python/ directory
from parquet_lab import metadata, prune, schema

path = "fixtures/pruning-sorted.parquet"  # try pruning-shuffled.parquet
data = open(path, "rb").read()
length = int.from_bytes(data[-8:-4], "little")
base = len(data) - 8 - length
md = metadata.decode_file_metadata(data[base:-8], base)
leaves = schema.leaves(schema.build(md.schema))
leaf = leaves[0]  # order_id; try 1, customer_id, with "424242"
converted = md.schema[leaf.element].converted_type
p = prune.Predicate.new(leaf, converted, prune.Op.EQ, "431")
use = prune.Mechanisms(statistics=True, bloom=True, page_index=True)  # turn each off
every = [x.column for x in leaves]  # SELECT *

plan = prune.plan(data, md, leaf, p, every, use)
for g in plan.row_groups:
    print(f"row group {g.index}: {'skip' if g.skipped else f'read rows {g.rows}'}")
    for mechanism, decision in g.steps:
        print(f"  {mechanism}: {'skip' if decision.skip else 'read'}, {decision.why}")
print(f"{plan.bytes_read()} bytes of column chunks, {plan.index_bytes()} to decide")

//! Plan SELECT * WHERE order_id = 431 with the reader, one mechanism at a time.

use parquet_lab::metadata::decode_file_metadata;
use parquet_lab::prune::{plan, Mechanisms, Op, Predicate};
use parquet_lab::schema::{build, leaves};

fn main() {
    let path = "fixtures/pruning-sorted.parquet"; // try pruning-shuffled.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 leaves = leaves(&build(&md.schema).expect("a schema"));
    let leaf = &leaves[0]; // order_id; try 1, customer_id, with "424242"
    let converted = md.schema[leaf.element].converted_type.as_deref();
    let p = Predicate::new(leaf, converted, Op::Eq, "431").expect("a condition");
    let use_ = Mechanisms {
        statistics: true,
        bloom: true,
        page_index: true, // turn each off
    };
    let every: Vec<usize> = leaves.iter().map(|x| x.column).collect(); // SELECT *

    let plan = plan(&data, &md, leaf, &p, &every, use_).expect("a plan");
    for g in &plan.row_groups {
        if g.skipped {
            println!("row group {}: skip", g.index);
        } else {
            println!("row group {}: read rows {:?}", g.index, g.rows);
        }
        for (mechanism, decision) in &g.steps {
            let verdict = if decision.skip { "skip" } else { "read" };
            println!("  {mechanism}: {verdict}, {}", decision.why);
        }
    }
    let (bytes, index) = (plan.bytes_read(), plan.index_bytes());
    println!("{bytes} bytes of column chunks, {index} to decide");
}

Statistics skip three row groups, as the first step did by hand. In row group 2, the page index keeps one page of five, and the reader reads that page’s rows from every column. The last line counts the bytes of column chunks the plan reads, and the bytes of page index it fetched to decide.

Set page_index=False (page_index: false in Rust): the reader keeps all of row group 2, and reads several times as many bytes. Then take leaves[1], which is customer_id, with "424242" on pruning-shuffled.parquet. Statistics keep every row group, and the Bloom filters skip every one, so the plan reads nothing but the filters. prune.Op has the other comparisons. Try amount_cents > 9800, which is leaves[3] with Op.GT (Op::Gt in Rust), or country is null, which is leaves[2] with Op.IS_NULL (Op::IsNull), whose value the reader ignores.

The same two conditions, with the mechanisms allowed one at a time:

Each mechanism costs bytes of its own: the page index and the filters are fetched before any data, and in an object store each fetch is a request. Row group statistics are free, because they came with the footer. Whether the others pay depends on how much they let the reader skip, and on how many requests they add, which ch10 measures.

Building it#

The steps read the statistics by hand, then asked the reader’s page_index and bloom modules, then called prune.plan. This section builds the parts of the plan: the test of a condition against a range, the probe of a filter, and the step from kept rows to the pages of every column. The tabs switch every excerpt on the page between the two languages.

A condition against a range#

against_bounds answers the null conditions first, from the null count alone, and skips a chunk whose values are all null for any comparison:

def against_bounds(
    p: Predicate, found: tuple[bytes, bytes] | None, null_count: int | None, num_values: int
) -> Decision:
    """Decide from a minimum and maximum (when usable), a null count, and a value count."""
    all_null = null_count == num_values and num_values > 0
    if p.op is Op.IS_NULL:
        return (
            skip("the null count is zero")
            if null_count == 0
            else read("the null count is not zero, or not recorded")
        )
    if p.op is Op.IS_NOT_NULL:
        return skip("every value is null") if all_null else read("some values are not null")
    if all_null:
        return skip("every value is null, and null satisfies no comparison")

pub fn against_bounds(
    p: &Predicate,
    bounds: Option<(&[u8], &[u8])>,
    null_count: Option<i64>,
    num_values: i64,
) -> Decision {
    let all_null = null_count == Some(num_values) && num_values > 0;
    match p.op {
        Op::IsNull if null_count == Some(0) => return skip("the null count is zero"),
        Op::IsNull => return read("the null count is not zero, or not recorded"),
        Op::IsNotNull if all_null => return skip("every value is null"),
        Op::IsNotNull => return read("some values are not null"),
        _ if all_null => return skip("every value is null, and null satisfies no comparison"),
        _ => {}
    }

For a comparison, against_bounds needs a usable minimum and maximum, which ch08’s bounds decided. It places the value against each, and rules the comparison out by the fact in the table above:

    if p.value is None or found is None:
        return read("no usable minimum and maximum")
    low, high = found
    vs_min, vs_max = p.comparator.compare(p.value, low), p.comparator.compare(p.value, high)
    if vs_min is None or vs_max is None:
        return read("the value cannot be placed in the column's order")
    ruled_out = {
        Op.EQ: vs_min == -1 or vs_max == 1,
        Op.NOT_EQ: vs_min == 0 and vs_max == 0,
        Op.LT: vs_min != 1,  # min >= x
        Op.LT_EQ: vs_min == -1,  # min > x
        Op.GT: vs_max != -1,  # max <= x
        Op.GT_EQ: vs_max == 1,  # max < x
    }[p.op]
    if not ruled_out:
        return read("the range may hold a match")
    if p.op is Op.EQ and vs_min == -1:
        return skip("the value is below the minimum")
    if p.op is Op.EQ:
        return skip("the value is above the maximum")
    if p.op is Op.NOT_EQ:
        return skip("every value equals it")
    if p.op in (Op.LT, Op.LT_EQ):
        return skip("the minimum is too large")
    return skip("the maximum is too small")

    let (Some(x), Some((min, max))) = (&p.value, bounds) else {
        return read("no usable minimum and maximum");
    };
    let c = p.comparator;
    let (vs_min, vs_max) = match (c.compare(x, min), c.compare(x, max)) {
        (Some(a), Some(b)) => (a, b),
        _ => return read("the value cannot be placed in the column's order"),
    };
    use Ordering::*;
    let ruled_out = match p.op {
        Op::Eq => vs_min == Less || vs_max == Greater,
        Op::NotEq => vs_min == Equal && vs_max == Equal,
        Op::Lt => vs_min != Greater,   // min >= x
        Op::LtEq => vs_min == Less,    // min > x
        Op::Gt => vs_max != Less,      // max <= x
        Op::GtEq => vs_max == Greater, // max < x
        Op::IsNull | Op::IsNotNull => unreachable!(),
    };
    if ruled_out {
        skip(match p.op {
            Op::Eq if vs_min == Less => "the value is below the minimum",
            Op::Eq => "the value is above the maximum",
            Op::NotEq => "every value equals it",
            Op::Lt | Op::LtEq => "the minimum is too large",
            _ => "the maximum is too small",
        })
    } else {
        read("the range may hold a match")
    }
}

The plan calls the same function in two places: with a column chunk’s statistics, and with each page’s entry in the ColumnIndex.

Probing a Bloom filter#

The block and the eight bits come from the hash alone:

    def probe_hash(self, hash: int) -> Probe:
        # The high 32 bits choose the block, scaled to the number of blocks without a
        # division.
        block = ((hash >> 32) * self.num_blocks()) >> 32
        # The low 32 bits, multiplied by each salt, choose one bit in each word: the top
        # five bits of the product are a bit position from 0 to 31.
        key = hash & 0xFFFF_FFFF
        bits = []
        for w, salt in enumerate(SALT):
            position = ((key * salt) & 0xFFFF_FFFF) >> 27
            at = 32 * block + 4 * w
            word = int.from_bytes(self.bitset[at : at + 4], "little")
            bits.append((position, bool(word >> position & 1)))
        return Probe(hash, block, bits, all(b for _, b in bits))

    pub fn probe_hash(&self, hash: u64) -> Probe {
        // The high 32 bits choose the block, scaled to the number of blocks without a
        // division.
        let block = (((hash >> 32) * self.num_blocks() as u64) >> 32) as usize;
        // The low 32 bits, multiplied by each salt, choose one bit in each word: the top
        // five bits of the product are a bit position from 0 to 31.
        let key = hash as u32;
        let mut bits = [(0u32, false); 8];
        for (w, bit) in bits.iter_mut().enumerate() {
            let position = key.wrapping_mul(SALT[w]) >> 27;
            let at = 32 * block + 4 * w;
            let word = u32::from_le_bytes(self.bitset[at..at + 4].try_into().unwrap());
            *bit = (position, word & (1 << position) != 0);
        }
        Probe {
            hash,
            block,
            bits,
            may_contain: bits.iter().all(|&(_, set)| set),
        }
    }
}

xxHash64 itself is about fifty lines in the same file, checked against the reference hashes of short strings.

From kept rows to the pages of every column#

A page’s rows run from its first row to the next page’s first row, or to the end of the row group:

    def row_ranges(self, num_rows: int) -> list[tuple[int, int]]:
        """The rows each page holds, as ``[first, end)``, given the row group's row count."""
        ends = [p.first_row_index for p in self.pages[1:]] + [num_rows]
        return [(p.first_row_index, end) for p, end in zip(self.pages, ends, strict=True)]

impl OffsetIndex {
    /// The rows each page holds, as `[first, end)`, given the row group's row count.
    pub fn row_ranges(&self, num_rows: i64) -> Vec<(i64, i64)> {
        (0..self.pages.len())
            .map(|i| {
                let end = self
                    .pages
                    .get(i + 1)
                    .map(|p| p.first_row_index)
                    .unwrap_or(num_rows);
                (self.pages[i].first_row_index, end)
            })
            .collect()
    }
}

column_read then takes the rows the condition’s column kept and one column chunk. With the chunk’s OffsetIndex, oi, it reads the dictionary page, if the chunk has one, and every data page whose rows overlap the kept rows. Without an OffsetIndex, it reads the whole chunk, whole:

    spans = []
    # The dictionary page sits before the first data page; any page read needs it.
    first_data = oi.pages[0].offset if oi.pages else whole.end
    ranges = oi.row_ranges(num_rows)
    wanted = [i for i in range(len(oi.pages)) if any(_overlaps(ranges[i], r) for r in rows)]
    if wanted and first_data > whole.start:
        spans.append(Span(whole.start, first_data))
    spans += [oi.pages[i].span() for i in wanted]
    return ColumnRead(column, spans, len(wanted), len(oi.pages)), oi.span.length

    let mut spans = Vec::new();
    // The dictionary page sits before the first data page; any page read needs it.
    let first_data = oi
        .pages
        .first()
        .map(|p| p.offset as u64)
        .unwrap_or(whole.end);
    let ranges = oi.row_ranges(num_rows);
    let wanted: Vec<usize> = (0..oi.pages.len())
        .filter(|&i| rows.iter().any(|&r| overlaps(ranges[i], r)))
        .collect();
    if !wanted.is_empty() && first_data > whole.start {
        spans.push(Span::new(whole.start, first_data));
    }
    spans.extend(wanted.iter().map(|&i| oi.pages[i].span()));
    (
        ColumnRead {
            column,
            spans,
            pages_read: wanted.len(),
            pages_total: oi.pages.len(),
        },
        oi.span.len(),
    )
}

Checking it#

Three kinds of test hold the reader to the data rather than to itself. The OffsetIndex must list exactly the pages the walker from ch06 finds. The ColumnIndex must hold each page’s own minimum, maximum and null count, as the reader computes them from the decoded values. And for hundreds of conditions on every fixture, every row that matches must lie in a row range the plan keeps, in a page every column reads. Every value in a chunk must also pass its Bloom filter.

python3 -m pytest python/tests

cargo test -p parquet-lab --test fixtures

Ask a library#

pyarrow and the parquet crate can both skip row groups. Ask each the questions the steps asked:

# The same questions to pyarrow: which row groups a condition keeps, and what else it sees.
import pyarrow.dataset as ds, pyarrow.parquet as pq

for path, column, value in [("fixtures/pruning-sorted.parquet", "order_id", 431),
                            ("fixtures/pruning-shuffled.parquet", "customer_id", 424242)]:
    fragment = next(ds.dataset(path).get_fragments())  # one file, one fragment
    kept = fragment.subset(ds.field(column) == value).row_groups  # by statistics alone
    print(f"{column} = {value}: kept row groups {[rg.id for rg in kept]}")

chunk = pq.read_metadata(path).row_group(0).column(1)  # customer_id, in the shuffled file
print(f"a column index: {chunk.has_column_index}; an offset index: {chunk.has_offset_index}")

//! The same questions to the `parquet` crate: row groups kept by a predicate you write on
//! their statistics, a page index, and Bloom filters, each read only when you ask for it.

use std::fs::File;

use parquet::file::metadata::RowGroupMetaData;
use parquet::file::page_index::column_index::ColumnIndexMetaData;
use parquet::file::properties::ReaderProperties;
use parquet::file::reader::{FileReader, SerializedFileReader};
use parquet::file::serialized_reader::ReadOptionsBuilder;
use parquet::file::statistics::Statistics;

/// Whether a row group's statistics for order_id leave room for 431.
fn may_hold_431(row_group: &RowGroupMetaData, _: usize) -> bool {
    match row_group.column(0).statistics() {
        Some(Statistics::Int64(s)) => match (s.min_opt(), s.max_opt()) {
            (Some(&min), Some(&max)) => (min..=max).contains(&431),
            _ => true,
        },
        _ => true, // no statistics to rule it out
    }
}

fn open(path: &str) -> File {
    File::open(path).expect("run this from the repository's root")
}

fn main() {
    let options = ReadOptionsBuilder::new()
        .with_predicate(Box::new(may_hold_431))
        .with_page_index()
        .build();
    let file = open("fixtures/pruning-sorted.parquet");
    let reader = SerializedFileReader::new_with_options(file, options).expect("a file");
    let md = reader.metadata();
    for row_group in md.row_groups() {
        println!("order_id = 431: kept row group {:?}", row_group.ordinal());
    }
    // The page index of the first row group kept: order_id's ColumnIndex.
    let page_index = md.page_index_for_row_group(0);
    let index = page_index.column_index(0).expect("a ColumnIndex");
    let order = index.get_boundary_order().expect("an order");
    if let ColumnIndexMetaData::INT64(pages) = index {
        let (min, max) = (pages.min_values(), pages.max_values());
        println!("  pages in {order:?} order: {min:?} to {max:?}");
    }

    let props = ReaderProperties::builder()
        .set_read_bloom_filter(true)
        .build();
    let options = ReadOptionsBuilder::new()
        .with_reader_properties(props)
        .build();
    let file = open("fixtures/pruning-shuffled.parquet");
    let reader = SerializedFileReader::new_with_options(file, options).expect("a file");
    for i in 0..reader.num_row_groups() {
        let row_group = reader.get_row_group(i).expect("a row group");
        let filter = row_group.get_column_bloom_filter(1); // customer_id's
        let may = filter.map(|f| f.check(&424242i64));
        println!("customer_id = 424242: row group {i}, may contain it: {may:?}");
    }
}

pyarrow’s dataset reader splits the file into row groups and keeps those whose statistics leave room for the condition. For order_id = 431 on the sorted file it keeps row group 2, as the first step did. For customer_id = 424242 on the shuffled file it keeps all four, where the reader’s Bloom filters ruled out every one. pyarrow skips by statistics only. It uses neither the Bloom filters nor the page index to skip, and its Python API cannot read either. It says whether a column chunk has a ColumnIndex and an OffsetIndex, and versions newer than the one the page loads (why the page’s is older) also say where a chunk’s Bloom filter is, but no more. Within a kept row group, it reads every page of the columns it needs.

The crate does nothing unless you ask. Its row group predicate is a function you write, here one that reads order_id’s minimum and maximum, and it keeps row group 2 as well. Asked for the page index, it returns the kept row group’s ColumnIndex: the order and the pages’ bounds the second step printed. Asked for the Bloom filters, its check says, for every row group, that 424242 is not there. The crate’s Arrow reader skips pages by the page index, with row selections and row filters. It needs the crate’s arrow feature, which this workspace does not build.

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 skipping_with_a_library

What this cannot tell you#

How conditions combine. The reader’s plan takes one condition. Each condition yields row ranges, and AND intersects them while OR joins them; a row group is skipped only when the combined ranges are empty.

How to skip within repeated columns. In a column that repeats, a page’s values are not its rows, and a reader needs the level histograms to map one to the other. This reader does not use the page index for such columns.

Whether skipping is faster. Skipping saves bytes and decoding. It can cost requests: a plan that reads five scattered pages makes five requests where a full scan could make one. ch10 weighs the two.

Other ways to skip. Some readers fetch a column’s dictionary page and look the value up there before reading any data page. This reader does not.

Key takeaways#

Problems#

Three, in exercises/python/skipping_data.py, or in Rust in exercises/src/skipping_data.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.

9.1 Can this be skipped? Decide from a range and a null count whether a page can hold a match for four comparisons. The test checks every page of the pruning fixtures: you must never skip a page holding a match, and must skip every page the metadata rules out.

9.2 Probe a Bloom filter. Given a filter’s bitset and a value’s hash, test the eight bits. The test compares your answer with the reader’s for thousands of customer numbers.

9.3 Your own condition. No test: the data is yours. Take a condition your queries use often and a file they run against, and run PYTHONPATH=python python3 -m parquet_lab skipping FILE COLUMN OP VALUE or cargo run -p pqlab -- skipping FILE COLUMN OP VALUE. Record how many row groups and bytes it skips, and with which mechanisms. Then rewrite the file sorted by that column and run it again. A good answer reports both, says what the sort cost the other columns’ compression, and names a condition the sort does not help.

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

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

cargo test -p exercises --test skipping_data -- --ignored

Where to go next#

• The page index is specified in the format repository’s PageIndex.md, and Bloom filters in BloomFilter.md.
• Split block Bloom filters come from Putze, Sanders and Singler, Cache-, Hash- and Space-Efficient Bloom Filters, in the ACM Journal of Experimental Algorithmics.
• xxHash64 is specified in the xxHash repository.
• ch10 turns these plans into requests against an object store.

Built from commit c32f48e.