ch10How readers read

What you build: The complete read path over the object store: footer, indexes and data in phases, with coalescing and concurrency.

The question#

Why does a reader issue exactly the requests it does?

ch09 decided which bytes a query needs. Knowing the bytes is not the same as fetching them well. Each request to an object store costs a round trip before the first byte arrives, and that cost does not shrink with the request. A reader that needs many small ranges can make one request for each, one large request that covers them all, or something between, and it can make several at once. This chapter builds the reader’s read path and measures those choices.

The experiment#

One query, request by request#

The query is ch09’s: SELECT * WHERE order_id = 431 on pruning-sorted.parquet. There the reader planned it from a copy of the file on disk. Here its scan runs the plan against the simulated object store from ch02, which logs every request, the connection that carried it, and when it started and ended. A Strategy says how to make the requests: how to find the footer and how much of the tail to read first, how close two ranges must be to merge into one request, and how many connections to use.

The step is a few lines of code. In Python, run it in the page, change it with Edit, and run it again. In Rust, open it in a Codespace, or run it at a desk with cargo run -p walkthroughs --bin read_with_a_strategy. It continues ch02’s open_with_the_reader, which stopped once the footer was read:

# The book's reader runs SELECT * WHERE order_id = 431, and the store logs each request.
import sys
sys.path.insert(0, "python")  # the book's reader, from the repository's python/ directory
from parquet_lab.object_store import NetworkModel
from parquet_lab.prune import Op
from parquet_lab.reader import FooterOptions, Head, SuffixRange
from parquet_lab.scan import Query, Strategy, scan

data = open("fixtures/pruning-sorted.parquet", "rb").read()
query = Query([0, 1, 2, 3], (0, Op.EQ, "431"))  # every column, where order_id = 431

# Try SuffixRange() and prefetch=8192; then coalesce_gap=4096; then connections=4.
footer = FooterOptions(Head(), prefetch=8)
strategy = Strategy(footer=footer, connections=1, coalesce_gap=None)
result = scan(data, "orders.parquet", query, strategy, NetworkModel())

for r in result.requests:
    when = f"{r.start_us // 1000:>3}-{r.end_us // 1000:>3} ms"
    print(f"{when} on {r.connection + 1}: {r.method} {r.range or ''} {r.why}")
fetched, ms = result.bytes_fetched, result.elapsed_us // 1000
print(f"{len(result.requests)} requests, {fetched} bytes, {ms} ms")
print("rows matching:", result.matches)

//! The book's reader runs SELECT * WHERE order_id = 431, and the store logs each request.

use parquet_lab::object_store::NetworkModel;
use parquet_lab::prune::{Mechanisms, Op};
use parquet_lab::reader::{FooterOptions, SizeSource};
use parquet_lab::scan::{scan, Query, Strategy};

fn main() {
    let data = std::fs::read("fixtures/pruning-sorted.parquet")
        .expect("run this from the repository's root");
    let query = Query {
        columns: vec![0, 1, 2, 3], // every column, where order_id = 431
        condition: Some((0, Op::Eq, "431".to_string())),
    };

    // Try SizeSource::SuffixRange and prefetch 8192; then Some(4096); then 4 connections.
    let strategy = Strategy {
        footer: FooterOptions {
            size: SizeSource::Head,
            prefetch: 8,
        },
        connections: 1,
        coalesce_gap: None,
        whole_chunks: false,
        mechanisms: Mechanisms::ALL,
    };
    let model = NetworkModel::default();
    let result = scan(&data, "orders.parquet", &query, strategy, model).expect("rows");

    for r in &result.requests {
        let when = format!("{:>3}-{:>3} ms", r.start_us / 1000, r.end_us / 1000);
        let (method, range) = (r.method.name(), r.range.as_deref().unwrap_or(""));
        println!("{when} on {}: {method} {range} {}", r.connection + 1, r.why);
    }
    let (fetched, ms) = (result.bytes_fetched, result.elapsed_us / 1000);
    println!(
        "{} requests, {fetched} bytes, {ms} ms",
        result.requests.len()
    );
    println!("rows matching: {:?}", result.matches);
}

The first three requests are ch02’s: the size, the trailer, the footer. Next come the indexes the plan needs, one request each: order_id’s ColumnIndex for row group 2, the one row group the statistics keep, and an OffsetIndex for each column. Last come the pages, one request each, with country’s dictionary page apart from its data page. Every request starts when the one before it ends. The bytes are a small share of the file, and they move in almost no time. The time is the latency, paid once per request.

Change the strategy one line at a time, and run the step after each change:

• Read more of the tail. Find the footer with SuffixRange() and prefetch=8192 (SizeSource::SuffixRange and prefetch: 8192 in Rust). The first request brings the trailer, the footer and the indexes together, because the indexes sit immediately before the footer. Every index request disappears.
• Merge ranges. Add coalesce_gap=4096 (Some(4096)). The pages of the four columns become one request, which also reads the bytes between them.
• Use more connections. Put the gap back to None and set connections=4. The pages go out side by side, each on a connection of its own, but none starts before the footer has arrived.
• Read everything. Set prefetch=65536, more than the file holds. The first request reads all of it, and there is nothing left to ask for.

Phases#

A reader cannot ask for everything at once, because it does not know what to ask for. The requests fall into phases, each needing the answers to the one before:

1. The footer. The trailer says where the footer starts, so finding it takes one or two requests after the size is known: ch02’s strategies.
2. The indexes. Which Bloom filters and page indexes to fetch depends on the footer: where they are, and which row groups the statistics have already ruled out.
3. The data. Which pages to fetch depends on the indexes.

Within a phase the requests are independent, and a reader with several connections sends them together. Between phases it must wait. Latency is paid at least once per phase, however many connections there are.

The phases on a timeline#

The panel runs the same query with the Rust reader and draws each request in the lane of the connection that carried it. Pick a strategy from the list; the list is the rows of the table under What each choice costs, below.

Try these:

1. … and pages by the page index. The step as you first ran it. Grey finds the footer, orange fetches the indexes, blue fetches the pages, and each bar waits for the one before.
2. HEAD, trailer, footer; every column chunk, then … skipping row groups by statistics. Without the page index there is no orange phase. The statistics drop three row groups’ chunks, and the blue phase shrinks with them.
3. … with an 8 KiB suffix read for the footer. The orange phase is gone: the tail read brought the indexes with the footer.
4. … merging ranges within 4 KiB. The blue phase is one bar.
5. … on four connections, without merging. The blue bars stack in four lanes, and the phase takes the time of the longest lane. The grey bar still comes first, alone.
6. One suffix read of 64 KiB: the whole file. One bar, and nothing after it.

What each choice costs#

The same query, with each change made in turn:

The first rows follow ch09: skipping row groups by statistics cuts both bytes and requests. Then the page index cuts bytes further but adds requests, one for each index and each page, and the query gets slower. Bytes were not the problem. At the latency the table’s note gives, the number of requests decides the time, and every row below that attacks it: a larger tail brings the indexes with the footer, merging turns neighbouring pages into one request, and connections run requests side by side.

For a file this small, the fastest strategy is the crudest: read all of it. That is a real answer: a reader that knows a file is small can fetch all of it before parsing anything.

Reading everything#

A query with no condition must read every column chunk. How long that takes depends on how the requests are made:

One request per column chunk, one at a time, pays the latency for every request in turn: the footer’s three, then one per column chunk. More connections overlap the column chunks’ requests, but the footer’s three stay in sequence, since each depends on the last. Merging chunks that touch needs no extra bytes at all: a row group’s chunks are written end to end, so the whole data region is one range.

When bytes matter#

Latency favours few large requests; bandwidth favours fetching only what is needed. Which wins depends on the network:

On a fast network with high latency, reading the whole file wins even though it moves many times the bytes. On a slow link with low latency, the careful plan wins, because moving the extra bytes takes longer than the round trips it saves. Try it in the step: pass NetworkModel(latency_us=1000, bandwidth_bytes_per_sec=1_000_000) (the same fields in Rust’s NetworkModel { .. }) and compare prefetch=8192 with prefetch=65536. Real files are far larger than this one, which moves the balance toward the careful plan: nobody reads a gigabyte to find one row. Readers therefore merge ranges up to a threshold, and choose the threshold from the latency and bandwidth they expect.

Only what was fetched#

The reader keeps a copy of the file that starts empty, and fills in each response. It plans from that copy and decodes from it. A range it forgot to request reads as zeros, which fails to decode. The tests run every query under every strategy and compare the rows with a plain read of the whole file, so a missed range cannot pass unnoticed.

Building it#

The step called scan, which reads the footer with ch02’s read_footer, plans with ch09’s prune.plan, and fetches in phases. This section builds the fetching: how ranges merge, how one phase is sent, how the store’s clock prices it, and how the reader decodes only the pages it fetched. The tabs switch every excerpt on the page between the two languages.

Merging ranges#

coalesce sorts the ranges and merges each into the one before it when the gap between them is at most gap bytes. With no gap it merges only ranges that overlap:

def coalesce(spans: list[Span], gap: int | None) -> list[Span]:
    """Sort ranges and merge those whose gap is at most ``gap`` bytes. With no gap, only ranges
    that overlap are merged, since fetching the same byte twice is never useful."""

    def merges(s: Span, last: Span) -> bool:
        return s.start <= last.end + gap if gap is not None else s.start < last.end

    out: list[Span] = []
    for s in sorted(spans):
        if out and merges(s, out[-1]):
            out[-1] = Span(out[-1].start, max(out[-1].end, s.end))
        else:
            out.append(s)
    return out

pub fn coalesce(mut spans: Vec<Span>, gap: Option<u64>) -> Vec<Span> {
    spans.sort();
    let mut out: Vec<Span> = Vec::new();
    for s in spans {
        let merges = |last: &Span| match gap {
            Some(g) => s.start <= last.end.saturating_add(g),
            None => s.start < last.end,
        };
        match out.last_mut() {
            Some(last) if merges(last) => {
                *last = Span::new(last.start, last.end.max(s.end));
            }
            _ => out.push(s),
        }
    }
    out
}

One phase#

fetch sends one phase. It drops the bytes the reader already holds, so a range the tail read brought is not fetched twice, merges what is left, and tells the store a new phase has begun:

def fetch(
    store: ObjectStore,
    key: str,
    have: Fetched,
    wanted: list[Span],
    gap: int | None,
    why: str,
) -> int:
    """Fetch every range the reader does not already have, coalesced, as one phase."""
    # Only the bytes not already held: a merged request must not fetch the tail again.
    missing = [part for s in wanted for part in have.missing(s)]
    planned = sum(s.length for s in coalesce(missing, 0))
    requests = [part for r in coalesce(missing, gap) for part in have.missing(r)]
    if not requests:
        return 0
    store.next_phase()
    for r in requests:
        got = store.get(key, Bounded(r), why)
        have.add(got.span, got.data)
    return planned

fn fetch<S: ObjectStore>(
    store: &mut S,
    key: &str,
    have: &mut Fetched,
    wanted: Vec<Span>,
    gap: Option<u64>,
    why: &str,
) -> Result<u64, String> {
    // Only the bytes not already held: a merged request must not fetch the tail again.
    let missing: Vec<Span> = wanted.into_iter().flat_map(|s| have.missing(s)).collect();
    let planned: u64 = coalesce(missing.clone(), Some(0))
        .iter()
        .map(|s| s.len())
        .sum();
    let requests: Vec<Span> = coalesce(missing, gap)
        .into_iter()
        .flat_map(|r| have.missing(r))
        .collect();
    if requests.is_empty() {
        return Ok(0);
    }
    store.next_phase();
    for r in requests {
        let got = store
            .get(key, GetRange::Bounded(r), why)
            .map_err(|e| e.to_string())?;
        have.add(got.span, &got.bytes);
    }
    Ok(planned)
}

scan calls it twice, once for the indexes and once for the pages, after read_footer has made the first phase’s requests.

A clock for every connection#

The simulated store gives each request the connection that is free first, and never starts it before its phase:

        # The connection that is free first; ties go to the lowest number.
        free, connection = min((t, i) for i, t in enumerate(self.free_at))
        start_us = max(free, self.phase_start)
        end_us = start_us + self.model.cost_us(nbytes)
        self.free_at[connection] = end_us

        // The connection that is free first; ties go to the lowest number.
        let (connection, free) = self
            .free_at
            .iter()
            .copied()
            .enumerate()
            .min_by_key(|&(i, t)| (t, i))
            .unwrap_or((0, 0));
        let start_us = free.max(self.phase_start);
        let end_us = start_us + self.model.cost_us(bytes);
        self.free_at[connection] = end_us;

Reading chosen pages#

With the page index, the reader parses each wanted page where the OffsetIndex says it starts, without walking the pages before it. The bytes between the pages were never fetched, and read_column_pages never looks at them:

def read_column_pages(
    file: bytes,
    chunk: ColumnChunk,
    leaf: Leaf,
    offsets: list[int],
) -> ColumnData:
    """Read only some pages of a column chunk: its dictionary page, if it has one, and the data
    pages starting at ``offsets``, each read where the page index says it is (ch10). The bytes
    between them are never looked at, so a reader need not have fetched them."""
    refuse_encrypted(chunk)
    r = chunk.byte_range()

    def one(at: int) -> Page:
        if at > r.end or at > len(file):
            raise ColumnError(f"no page at offset {at} in chunk {r}")
        try:
            return read_page(ByteReader(file[at : r.end], at))
        except ValueError as e:
            raise ColumnError(f"page at offset {at}: {e}") from e

    pages = []
    if chunk.dictionary_page_offset is not None and chunk.dictionary_page_offset > 0:
        pages.append(one(r.start))
    pages += [one(at) for at in offsets]
    return decode_pages(file, chunk, leaf, pages)

pub fn read_column_pages(
    file: &[u8],
    chunk: &ColumnChunk,
    leaf: &Leaf,
    offsets: &[u64],
) -> Result<ColumnData, ColumnError> {
    refuse_encrypted(chunk)?;
    let range = chunk.byte_range();
    let one = |at: u64| -> Result<Page, ColumnError> {
        let Some(bytes) = file.get(at as usize..range.end as usize) else {
            return err(format!("no page at offset {at} in chunk {range}"));
        };
        crate::pages::read_page(&mut ByteReader::new(bytes, at))
            .map_err(|e| ColumnError(format!("page at offset {at}: {e}")))
    };
    let mut pages = Vec::new();
    if chunk.dictionary_page_offset.is_some_and(|d| d > 0) {
        pages.push(one(range.start)?);
    }
    for &at in offsets {
        pages.push(one(at)?);
    }
    decode_pages(file, chunk, leaf, pages)
}

Checking it#

Over two hundred combinations of fixture, condition and strategy, the reader’s rows must equal a plain read’s:

python3 -m pytest python/tests -k every_strategy

cargo test -p parquet-lab --test fixtures every_strategy

Ask a library#

A library makes the same choices, with its own defaults. Hand pyarrow a file that prints every read it is asked for, and ask the parquet crate for the footer and the page index from a tail of the file:

# pyarrow reads row group 2 of the same file, from a file that prints each read it gets.
import io
import pyarrow.parquet as pq

class Logged(io.RawIOBase):
    def __init__(self, path): self.data, self.at = open(path, "rb").read(), 0
    def readable(self): return True
    def seekable(self): return True
    def tell(self): return self.at
    def seek(self, offset, whence=0):
        self.at = [offset, self.at + offset, len(self.data) + offset][whence]
        return self.at
    def readinto(self, buffer):
        n = min(len(buffer), len(self.data) - self.at)
        buffer[:n] = self.data[self.at : self.at + n]
        print(f"  read {n} bytes at {self.at}")
        self.at += n
        return n

for pre_buffer in (False, True):  # pyarrow's default is True
    file = Logged("fixtures/pruning-sorted.parquet")
    print(f"open, pre_buffer={pre_buffer}:")
    parquet = pq.ParquetFile(file, pre_buffer=pre_buffer)
    print("read row group 2:")
    rows = parquet.read_row_groups([2]).num_rows  # try pq.read_table(file, filters=...)
    print(f"{rows} rows")

//! The `parquet` crate reads the same file's tail, and asks for more until it has enough.

use bytes::Bytes;
use parquet::errors::ParquetError;
use parquet::file::metadata::{PageIndexPolicy, ParquetMetaDataReader};

fn main() {
    let data = std::fs::read("fixtures/pruning-sorted.parquet")
        .expect("run this from the repository's root");
    let size = data.len() as u64;
    let policy = PageIndexPolicy::Required; // the footer, and the page index before it
    let mut reader = ParquetMetaDataReader::new().with_page_index_policy(policy);

    let mut tail = 8; // try 8192, or 65536
    loop {
        println!("read the last {tail} bytes");
        let bytes = Bytes::copy_from_slice(&data[data.len().saturating_sub(tail)..]);
        match reader.try_parse_sized(&bytes, size) {
            Ok(()) => break,
            Err(ParquetError::NeedMoreData(needed)) => {
                println!("  NeedMoreData({needed})");
                tail = needed;
            }
            Err(e) => panic!("{e}"),
        }
    }
    let md = reader.finish().expect("the footer and the page index");
    let (groups, index) = (md.num_row_groups(), md.page_index().is_some());
    println!("{groups} row groups; the page index read: {index}");
}

To open the file, pyarrow reads a fixed amount from the end, 64 KiB by default. This file is smaller, so that first read is the whole file. pyarrow keeps only the footer from it, and reading row group 2 reads the row group’s column chunks again.

With pre_buffer=False, pyarrow makes one read per column chunk, the whole chunk each time, since it does not use the page index. With pre_buffer=True, its default, it merges the chunks, which touch, into one read. The gap it merges across is hole_size_limit in pyarrow.CacheOptions, which a dataset scan takes through ParquetFragmentScanOptions(cache_options=...). The page runs an older build of pyarrow for WebAssembly (why), which reads a chunk at a time either way; run the step at a desk to see the merged read.

In place of read_row_groups, try pq.read_table(file, filters=[("order_id", "==", 431)]). The statistics skip three row groups, and the fourth is read whole: pyarrow does not use the page index to skip pages.

The crate’s ParquetMetaDataReader makes no requests of its own. Given the last eight bytes, it answers NeedMoreData with the size of tail it needs: the footer and its trailer. Given that, it parses the footer, finds the page index before it, and asks again, for a tail reaching back to the first index. Given that, it has everything. Those are the chapter’s first two phases, with each answer needed before the next request can be made. Start from 8192 and it needs one read. The crate’s asynchronous reader takes the same number through with_prefetch_hint, and hands a row group’s ranges to the object_store crate’s get_ranges, which merges ranges that lie close together. It needs the crate’s arrow and async features, 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 reads_with_a_library

What this cannot tell you#

How a real store behaves. The model charges a fixed latency and a fixed bandwidth. Real stores vary from request to request, throttle busy prefixes, and limit connections. The model keeps the shape of the problem and none of its noise, so its numbers compare strategies; they do not predict a real query’s time.

Caching. A reader that opens the same file twice can keep its footer, and one that runs many queries can keep pages. This reader starts empty every time.

Reading many files. A table is usually many files, and a query engine reads them in parallel. ch14 turns to tables.

Decoding time. Every cost here is the network’s. On a fast network, decompressing and decoding can take longer than fetching, and a reader that overlaps the two gains most.

Key takeaways#

Problems#

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

10.1 Merge ranges. Turn byte ranges into requests, merging those within a gap. The test compares your requests with the reader’s on thousands of generated cases.

10.2 Time the requests. Given each phase’s request durations and a number of connections, say when the last request finishes. The test compares your answer with the simulated store’s clock.

10.3 Your own network. No test: the network is yours. Time a hundred small range requests and a few large ones against the object store your files live in, and estimate its latency and bandwidth. Then run PYTHONPATH=python python3 -m parquet_lab scan FILE --where ... --latency-us N --bandwidth N or cargo run -p pqlab -- scan FILE --where ... --latency-us N --bandwidth N on one of your files, with a few gaps and connection counts. A good answer names the gap and connection count you would choose, and checks the prediction against a real query’s timing.

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

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

cargo test -p exercises --test how_readers_read -- --ignored

Where to go next#

• Amazon’s guidance on range requests and parallel connections describes the behaviour this chapter models.
• Production readers expose these choices as options. The Arrow project’s Rust Parquet reader, for one, takes a hint for how much of the tail to read for the footer, and merges nearby ranges.
• ch11 turns from reading to writing: how a writer’s choices decide what any reader can skip.

Built from commit c32f48e.