ch11Writing Parquet well

What you build: The same rows written seven ways, each measured by what the reader must fetch to query it.

The question#

Which writer settings decide how cheap a file is to read?

Every chapter since ch08 has been about what a reader can avoid. It can avoid only what the file lets it. Row groups, their order, pages, encodings and indexes are all fixed when the file is written, and no reader can undo a writer’s choice. This chapter writes the same rows seven ways and measures each by what queries against it cost.

The experiment#

Seven files, one table#

The writing-*.parquet files hold the same orders. writing-baseline.parquet is sorted by order_id, dictionary-encoded, compressed with Snappy, and written in four row groups of small pages with a page index. Each other file changes one of those settings: one row group, many small ones, sorted by country, shuffled, no dictionary, or no page index. Appendix B lists 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.

Row groups a lookup reads. The query is order_id = 431. A reader reads every row group whose footer range could hold the value. Read each file’s footer, as ch09 did, and count those row groups:

# Read each file's footer: how many of its row groups could 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

files = "baseline one-group small-groups by-country shuffled plain no-index".split()
column, wanted = 0, 431  # order_id = 431; try 2, customer_id, and 500000

for name in files:
    data = open(f"fixtures/writing-{name}.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)
    hits = 0
    for row_group in md.row_groups:
        s = row_group.columns[column].statistics  # an INT64: eight bytes, least first
        low = int.from_bytes(s.min_value, "little", signed=True)
        high = int.from_bytes(s.max_value, "little", signed=True)
        hits += low <= wanted <= high
    groups = len(md.row_groups)
    print(f"writing-{name}: footer {length} bytes, {hits} of {groups} row groups may hold it")

//! Read each file's footer: how many of its row groups could hold order 431?

use parquet_lab::metadata::decode_file_metadata;

fn main() {
    let files = "baseline one-group small-groups by-country shuffled plain no-index";
    let (column, wanted) = (0, 431); // order_id = 431; try 2, customer_id, and 500000

    for name in files.split_whitespace() {
        let path = format!("fixtures/writing-{name}.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 mut hits = 0;
        for row_group in &md.row_groups {
            let chunk = &row_group.columns[column]; // an INT64: eight bytes, least first
            let s = chunk.statistics.as_ref().expect("statistics");
            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));
            hits += usize::from((low..=high).contains(&wanted));
        }
        let groups = md.row_groups.len();
        println!(
            "writing-{name}: footer {length} bytes, {hits} of {groups} row groups may hold it"
        );
    }
}

Every file sorted by order_id sends the lookup to one row group, however many row groups it has. The file sorted by country and the shuffled one send it to all of theirs, because each of their row groups spans nearly every order number. The footers differ too: the file of small row groups has one many times larger than the file of one row group.

Change column to 2, which is customer_id, and wanted to 500000. Nobody sorted by customer, so every file sends that lookup to every row group it has. Small row groups do not help a column nobody sorted by; they only add row groups to read.

One value can be lucky. Averaged over every value of the two columns, the reader counted:

What a query reads#

One query, every file. The book’s reader from ch10 runs SELECT * WHERE order_id = 431 against each file in turn. It finds the footer exactly, uses every skipping mechanism, and merges only ranges that touch. What it reads after the footer is then what the file’s layout made it read:

# The book's reader runs one query on every file: what does it read after the footer?
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
from parquet_lab.scan import Query, Strategy, scan

files = "baseline one-group small-groups by-country shuffled plain no-index".split()
# Every column, where order_id = 431; try (3, Op.EQ, "FR"), then (4, Op.EQ, "refunded").
query = Query([0, 1, 2, 3, 4, 5], (0, Op.EQ, "431"))
# The footer found exactly, every skipping mechanism, and only ranges that touch merged.
strategy = Strategy(footer=FooterOptions(Head(), prefetch=8), coalesce_gap=0)
later = ("read the indexes", "read the pages")  # the requests after the footer

for name in files:
    data = open(f"fixtures/writing-{name}.parquet", "rb").read()
    result = scan(data, name, query, strategy, NetworkModel())
    after = [r for r in result.requests if r.why.startswith(later)]
    fetched, rows = sum(r.bytes_returned for r in after), len(result.matches)
    print(f"writing-{name}: {fetched} bytes in {len(after)} requests, {rows} rows matching")

//! The book's reader runs one query on every file: what does it read after the footer?

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 files = "baseline one-group small-groups by-country shuffled plain no-index";
    let query = Query {
        columns: vec![0, 1, 2, 3, 4, 5], // every column, where order_id = 431
        // Try 3, country, and "FR"; then 4, status, and "refunded".
        condition: Some((0, Op::Eq, "431".to_string())),
    };
    // The footer found exactly, every skipping mechanism, and only ranges that touch merged.
    let strategy = Strategy {
        footer: FooterOptions {
            size: SizeSource::Head,
            prefetch: 8,
        },
        connections: 1,
        coalesce_gap: Some(0),
        whole_chunks: false,
        mechanisms: Mechanisms::ALL,
    };
    let later = ["read the indexes", "read the pages"]; // the requests after the footer

    for name in files.split_whitespace() {
        let path = format!("fixtures/writing-{name}.parquet");
        let data = std::fs::read(path).expect("run this from the repository's root");
        let model = NetworkModel::default();
        let result = scan(&data, name, &query, strategy, model).expect("rows");
        let after: Vec<_> = (result.requests.iter())
            .filter(|r| later.iter().any(|w| r.why.starts_with(w)))
            .collect();
        let fetched: u64 = after.iter().map(|r| r.bytes_returned).sum();
        let (n, rows) = (after.len(), result.matches.len());
        println!("writing-{name}: {fetched} bytes in {n} requests, {rows} rows matching");
    }
}

Every file returns the same order. What it cost differs:

• Sorting decides. The file sorted by country and the shuffled one read most of their bytes, as the first step predicted. The small row groups read least, since the one row group they read is small.
• Dictionaries make point lookups dearer. To decode one page of a dictionary-encoded column, the reader must first read that column’s dictionary page, which for a column of unique values holds every value in the row group. The baseline read several times more than the file without dictionaries.
• The page index trades bytes for requests. The file without one read its whole row group in one request. The baseline read fewer bytes in more requests. Against object storage, as ch10 measured, the requests can cost more than the bytes saved.

Change the condition to country = 'FR', with column 3 and "FR": now the file sorted by country reads least. Then try status = 'refunded', with column 4. No file does well. Refunds are rare and spread through every row group, and no single sort order serves every column. The reader ran those conditions, and one on amounts, against every file:

Sorting makes lookups cheap only for the column sorted by. A rare value spread through every row group, such as a refund, costs nearly a full read whatever the layout.

What each setting changed#

The files themselves, before any query:

Every setting has a price somewhere:

• Row group size. Each row group repeats the metadata for every column: a column chunk entry in the footer, statistics, page index entries, and a dictionary page per column. The first step printed the footer of the small row groups many times larger than the footer of one row group, and the file is larger with it.
• Dictionary encoding. It shrank country, which repeats a few values, and grew order_id, whose values are all different: ch05’s lesson, at the scale of a file.
• Sort order. Sorting by country gathered each country’s rows together, and the country column shrank, because its dictionary indices now repeat in long runs.
• The page index. Turning it off made the column chunks larger, not smaller. pyarrow writes each page’s statistics into the page index when there is one, and into every page header when there is not.

Choosing#

No setting is right for every table, but the measurements point one way:

• Sort by the column queries filter on most, and expect other columns to gain little from statistics.
• Make row groups large. Small row groups multiply the footer and the per-chunk overhead, and help only a sorted column. Production writers commonly aim for tens to hundreds of megabytes per row group.
• Let dictionary encoding fall back for columns of mostly distinct values, or turn it off for them. Writers fall back when a dictionary grows too large, and a column of unique identifiers gains nothing from one.
• Write the page index for tables queried selectively on a sorted column, and expect it to help most when requests are cheap.
• Add Bloom filters for equality lookups on unsorted, high-cardinality columns, as ch09 showed for customer numbers.

Building it#

The steps read each footer by hand, then called the reader’s scan, which ch10 built. The reader is the same for every file; only the files differ. This section shows the settings the files were written with, and the two places in the reader where a file’s layout turns into bytes read: the indexes a lookup fetches, and the dictionary page it cannot skip.

The writer’s settings#

The fixtures are written by pyarrow, never by this repository’s code. These are the settings every writing-* file starts from:

WRITING_OPTIONS = {
    "compression": "snappy",
    "use_dictionary": True,
    "row_group_size": 200,
    "max_rows_per_page": 40,
    "write_page_index": True,
}

The indexes a lookup fetches#

Before it reads any data, scan fetches the indexes its plan will consult. A row group the footer’s statistics rule out needs none. For every other row group, it fetches the condition column’s Bloom filter if the writer wrote one, that column’s ColumnIndex, and the OffsetIndex of every column the query touches:

    # Phase 2: the indexes the plan will consult. Which ones depends on the footer alone.
    planned = 0
    if predicate is not None:
        leaf, p = predicate
        orders = md.column_orders
        type_order = orders is not None and leaf.column < len(orders) and orders[leaf.column] == "TYPE_ORDER"
        wanted = []
        for rg in md.row_groups:
            chunk = rg.columns[leaf.column]
            # A row group the footer's statistics already rule out needs no index.
            if strategy.mechanisms.statistics and chunk.statistics is not None:
                st = chunk.statistics
                try:
                    b = stats.bounds(st, p.comparator, type_order)
                    found = (b.min, b.max)
                except ValueError:
                    found = None
                if prune.against_bounds(p, found, st.null_count, chunk.num_values).skip:
                    continue
            if strategy.mechanisms.bloom and p.op is Op.EQ:
                o, n = chunk.bloom_filter_offset, chunk.bloom_filter_length
                if o is not None and n is not None:
                    wanted.append(Span(o, o + n))
            if strategy.mechanisms.page_index and not strategy.whole_chunks:
                if chunk.column_index is not None:
                    wanted.append(chunk.column_index)
                wanted += [rg.columns[c].offset_index for c in touched if rg.columns[c].offset_index]
        planned += fetch(store, key, have, wanted, strategy.coalesce_gap, "read the indexes the plan needs")

    // Phase 2: the indexes the plan will consult. Which ones depends on the footer alone.
    let mut planned = 0;
    if let Some((leaf, p)) = &predicate {
        let mut wanted = Vec::new();
        let type_order = md
            .column_orders
            .as_ref()
            .and_then(|o| o.get(leaf.column))
            .is_some_and(|o| o == "TYPE_ORDER");
        for rg in &md.row_groups {
            let chunk = &rg.columns[leaf.column];
            // A row group the footer's statistics already rule out needs no index.
            if strategy.mechanisms.statistics {
                if let Some(st) = &chunk.statistics {
                    let b = crate::stats::bounds(st, p.comparator, type_order).ok();
                    let bounds = b.as_ref().map(|b| (&b.min[..], &b.max[..]));
                    if prune::against_bounds(p, bounds, st.null_count, chunk.num_values).skip {
                        continue;
                    }
                }
            }
            if strategy.mechanisms.bloom && p.op == Op::Eq {
                if let (Some(o), Some(l)) = (chunk.bloom_filter_offset, chunk.bloom_filter_length) {
                    wanted.push(Span::new(o as u64, (o + l) as u64));
                }
            }
            if strategy.mechanisms.page_index && !strategy.whole_chunks {
                wanted.extend(chunk.column_index);
                wanted.extend(touched.iter().filter_map(|&c| rg.columns[c].offset_index));
            }
        }
        planned += fetch(
            store,
            key,
            &mut have,
            wanted,
            strategy.coalesce_gap,
            "read the indexes the plan needs",
        )?;
    }

A file sorted by the condition’s column leaves one row group, and so fetches one row group’s indexes. A shuffled file leaves every row group and fetches every row group’s indexes, before it reads a page. A file without a page index has nothing to fetch here, and reads whole column chunks instead.

The dictionary page a lookup cannot skip#

The plan then reads the pages of each column that hold the kept rows. ch09’s column_read adds the chunk’s dictionary page to any read of its data pages, since a dictionary-encoded page cannot be decoded without it:

    # 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))

    // 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));
    }

For country, the dictionary page is a few values. For order_id, whose values are all different, it is every value in the row group, which is why the lookup read more from the dictionary-encoded baseline than from the plain file.

Checking it#

Every file reassembles to the same rows, and every query returns the same matches from every file, whatever the layout:

python3 -m pytest python/tests

cargo test -p parquet-lab --test fixtures

Ask a library#

At work you choose these settings in a writer, and read them back from the footer. pyarrow writes the baseline’s rows again with the fixtures’ settings, then with one changed at a time. The parquet crate reads the choices back from each file’s footer:

# pyarrow writes the orders again, one setting changed at a time, and reads its footer back.
import io
import pyarrow.parquet as pq

table = pq.read_table("fixtures/writing-baseline.parquet")
# The fixtures' settings, in fixtures/generate.py, with no Arrow schema stored in the footer.
# A page ends at every size check, so a check every 40 rows makes pages of 40 rows: what
# max_rows_per_page=40 says in pyarrow 25, which the page's pyarrow predates.
baseline = dict(compression="snappy", use_dictionary=True, row_group_size=200,
                write_batch_size=40, data_page_size=1,
                write_page_index=True, store_schema=False)
changes = {
    "the baseline": {},
    "row groups of 40": {"row_group_size": 40},
    "no dictionary": {"use_dictionary": False},
    "no page index": {"write_page_index": False},
    "the sort declared": {"sorting_columns": [pq.SortingColumn(0)]},  # sorted by order_id
}
for change, options in changes.items():
    sink = io.BytesIO()
    pq.write_table(table, sink, **{**baseline, **options})
    md = pq.read_metadata(io.BytesIO(sink.getvalue()))
    size, footer, groups = len(sink.getvalue()), md.serialized_size, md.num_row_groups
    print(f"{change}: {size} bytes, footer {footer}, {groups} row groups")
    group = md.row_group(0)
    c = group.column(0)  # order_id in the first row group
    print(f"  dictionary {c.has_dictionary_page}, column index {c.has_column_index}, "
          f"sorted by {group.sorting_columns}")

//! The `parquet` crate reads each file's footer and finds the writer's choices in it.

use parquet::file::metadata::ParquetMetaDataReader;

fn main() {
    let files = "baseline one-group small-groups by-country shuffled plain no-index";
    for name in files.split_whitespace() {
        let path = format!("fixtures/writing-{name}.parquet");
        let file = std::fs::File::open(path).expect("run this from the repository's root");
        let md = ParquetMetaDataReader::new()
            .parse_and_finish(&file)
            .expect("a Parquet file");
        let group = md.row_group(0);
        let chunk = group.column(0); // order_id in the first row group
        let (groups, codec) = (md.num_row_groups(), chunk.compression());
        println!("writing-{name}: {groups} row groups, {codec}");
        let dictionary = chunk.dictionary_page_offset().is_some();
        let index = chunk.column_index_offset().is_some();
        let sorted = group.sorting_columns();
        println!("  dictionary {dictionary}, column index {index}, sorted by {sorted:?}");
    }
}

With the fixtures’ settings, pyarrow writes writing-baseline.parquet again, to the byte: the same size and the same footer. The pyarrow the page loads is older than the one that wrote the fixtures (why), and has no max_rows_per_page, so the step cuts its pages another way, as its comment says. Each change shows in the footer as the steps found it: more row groups and a larger footer, no dictionary page, no ColumnIndex. The last change declares the sort. sorting_columns records in each row group that it is sorted by order_id, for a few bytes of footer. The fixtures do not declare it, so the crate finds no sort order in any of them. A reader can infer the order from the statistics, but the footer never states it.

The crate writes too. Its WriterProperties set the same choices: set_max_row_group_row_count, set_dictionary_enabled, set_statistics_enabled with EnabledStatistics::Page for the page index, set_sorting_columns and set_column_bloom_filter_enabled. Writing rows with it needs either a column writer fed value by value, or the arrow feature’s ArrowWriter, which this workspace does not build, so the step reads instead.

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 writing_with_a_library

What this cannot tell you#

How the settings behave at scale. The fixtures hold few enough rows that every row group is small, and the proportions change with size. A footer that is a large share of a small file, as the first step printed, is a rounding error in a large one. The direction of each effect holds; the sizes do not.

What the settings cost the writer. Larger row groups need more memory while writing, and sorting needs the rows first. This chapter measures only readers.

Other writers. pyarrow’s defaults, and the way it places statistics, are pyarrow’s. Other writers choose differently, and a file’s created_by says which one wrote it.

Key takeaways#

Problems#

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

11.1 Row groups per lookup. From each row group’s bounds, count the row groups a lookup for a value must read. The test compares your count with the reader’s plans for values in every file.

11.2 What order costs. From a column’s values in file order and a row group size, compute the mean number of row groups a lookup reads. The test checks every ch11 file’s integer columns.

11.3 Your own writer. No test: the data is yours. Rewrite one of your files three ways, changing one setting each time: the sort column, the row group size, or dictionary encoding for one column. Measure each with PYTHONPATH=python python3 -m parquet_lab scan FILE --where ... or cargo run -p pqlab -- scan FILE --where ... on a query you run often. A good answer reports the file sizes, what each query read, and which setting you would change in production, with the cost that change puts on writing.

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

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

cargo test -p exercises --test writing_parquet_well -- --ignored

Where to go next#

• pyarrow’s write_table documents every setting used here.
• The format’s configuration notes discuss row group and page sizes.
• ch12 asks what a query engine does with a file like these, and what each kind of query costs against one.

Built from commit c32f48e.