One Number, Two Jobs

TL;DR

• Opteryx used to write Parquet row groups of about 256k rows. That size was picked so reads from object storage needed fewer range requests. It also set the morsel size, the unit of work the engine handles, and on TPC-H, 256k is not the best morsel size.
• Writing smaller row groups the normal way makes remote reads worse. Every projected column costs one range GET per row group, so 64k row groups mean about 3.6× the requests.
• We changed how Rugo lays files out. Row groups are now 64k rows, written in blocks of four. Inside a block, each column's four chunks sit next to each other, so the reader fetches 256k rows of a column in one range. On the ClickBench query set that's 5.8% fewer GETs than before, and the morsels are a quarter of the size.
• The files are still standard Parquet. PyArrow, DuckDB, Polars, Rugo and Opteryx all read them and return the same values.

The number that was doing two jobs

In a Parquet file, the row group is the basic unit. Each column in a row group is stored as one contiguous column chunk. The footer records min/max statistics for every chunk, and the reader uses those to decide whether it needs the chunk at all.

In Opteryx, the row group is also the morsel: the batch of rows that a worker picks up, decodes and pushes through the query plan. One row group turns into one morsel.

So one write-time setting controlled two separate things:

1. The IO cost. The remote reader turns each row group into one work item and merges range requests only within that row group. Bigger row groups mean fewer, larger GETs against GCS or S3.
2. The engine cost. A row group's size is the morsel's size. That decides how well the working set fits in CPU cache, how often we pay per-morsel dispatch overhead, and how evenly work spreads across workers.

We'd set it to 256k for the first reason. Back in July, Rugo's default was 262,144 rows. That's a good size for object storage, but we didn't know whether it was a good size for the engine.

What size does the engine want?

We measured it. We rewrote TPC-H SF1 and SF10 at 8k, 16k, 32k, 64k and 256k rows per row group and kept the source file boundaries. Then we ran all 22 queries with the arms interleaved and their order rotated each round. Results were identical across every size.

64k was fastest in every round of every run. Below 32k, per-morsel overhead dominates. The extra dispatching costs more than the better cache use saves. At 256k we lose a few percent on SF1 and break even on SF10.

The per-operator breakdown explains it. Aggregation is fastest at 32k on both scales and 3–8% slower at 256k, which is a cache effect. Scans are fastest at 64k. Inner joins are the one operator that likes big morsels: at SF10 on a single worker, joins ran 9% faster at 256k. The total still favours 64k.

So the engine wants 64k and object storage wants 256k.

Smaller row groups, done the obvious way

The obvious fix is to write 64k row groups and accept the cost. The cost is large.

A normal Parquet file is laid out row-major by row group. All of row group 1's column chunks come first, then all of row group 2's, and so on (the top half of the figure below).

If you're reading just c1, its chunks are spread through the file with other columns in between. Each one is a separate range request. Cutting row groups to a quarter of the size gives you four times as many chunks and roughly four times as many GETs.

We measured this by serving all 99 ClickBench hits files (99 million rows) over a local HTTP server that counts every range request. We then ran the 43 ClickBench statements through the native scan path with production settings:

On GCS, each request costs over 100 ms of round-trip time. Spending 3.5× the requests to get 4% faster on CPU is a bad deal.

Moving the chunks, not the boundaries

The solution is to stop requiring that a row group's chunks be stored together.

The Parquet spec doesn't require that. The footer records a byte offset and length for every column chunk separately. A row group is a logical grouping, and its chunks can be anywhere in the file. So Rugo now writes row groups in blocks of four (256k rows). Inside a block, each column's chunks are placed next to each other:

To read c1 across the block, you need one range instead of four. The row group stays exactly what it was before: the unit for decoding, statistics, pruning and morsels. The only change is where the bytes are, which changes how many requests it takes to fetch them.

On the reader side, a remote read is now a fetch block: the row groups that survived pruning in one block. They're planned together through the existing range coalescer and fetched as one batch. Then they're decoded and handed to the engine one row group at a time. Workers still get 64k-row morsels, but the network sees requests for 256k rows.

Same benchmark, same counting server:

The grouped layout uses slightly fewer requests than the old 256k files. We modelled this before writing any code, and the model's request counts came within 5% of the measured ones on all three layouts.

Why blocks of four?

If putting four row groups' chunks together saves requests, why not eight? We tried it. On the same ClickBench set, blocks of eight cut range GETs by a further 45% (20,995 to 11,481) and read exactly the same bytes. But production reads are limited by bandwidth more than by request count.

Modelled against GCS, eight saves 0.7% IO wait time but bigger blocks also cost something: a block's morsels are released only once the whole block has arrived, so the first one waits longer; LIMIT queries over-fetch more; and fetch-ahead, which was tuned on 256k-row fetch units, would switch on only for scans twice as large.

Four is the balance: units of IO small enough for effective work scheduling, and large enough to amortise the round trip. That balance depends on bandwidth and on how fetch-ahead is tuned, so if either changes, it's a setting to revisit, and because it's set when files are written, revisiting it only means rewriting files.

Finer pruning

Smaller row groups also mean finer statistics. A 256k row group with one matching row has to be read in full. At 64k, the three quarters of the block without a match can be skipped.

Q20 on ClickBench is a point lookup on a scattered UserID. It dropped from 592 GETs and 199 MB to 348 GETs and 133 MB. When a block is partly pruned, the reader fetches only the surviving row groups, still as one range when they're next to each other.

The reader works it out from offsets

We didn't add any new metadata. The reader works out blocks from the footer: two neighbouring row groups belong to the same block when every projected column's chunk in the second one starts exactly where its chunk in the first one ends.

That means:

• Files from other writers, and Rugo files written before this change, read the same way as before: one block per row group, with no special case.
• Blocks are decided per query, from the projected columns, so there's no flag that can disagree with the actual bytes.
• Footer fields stay accurate. RowGroup.file_offset is the start of the row group's earliest chunk, and total_compressed_size is the sum of its chunks.

The rule we set for the writer still applies: a file only Opteryx can read is a defect. The test suite writes grouped files with nine column types, including nulls, decimals, dictionary and plain strings, and with a page index. It reads them back through PyArrow, DuckDB, Polars, Rugo and Opteryx, full, projected and filtered, and compares the values.

What it cost us

Bloom filters moved to the end of the file. Before, a column's bloom filter sat right next to its chunk, so it came along in the same GET, and we probed it to skip decoding chunks with no match. In the grouped layout, blooms placed between chunks would break the adjacency, and blocks would stop merging across columns. With blooms at the tail, the model gave 23,943 GETs. With blooms in front of each block, it gave 28,589. So blooms now go after the last block, and we removed the remote bloom decode-skip. It only ever saved CPU, never bytes, and fetching blooms from the tail would add requests. Plan-time bloom pruning on cached footers is unchanged, and so is the dictionary decode-skip.

Files are about 6% bigger. Smaller row groups mean more dictionaries, more statistics and more blooms. On hits, 64k files average 95 bytes per row, compared with 89.5 at 256k. Long strings like URL and Title grow the most. That accounts for the +9.8% bytes above.

Footers are about 3.5× bigger. Footer size grows with the number of row groups, and the layout doesn't change that: about 6.2 MiB per 4 GiB file at 64k, against 1.8 MiB at 256k. Footers are read at plan time and cached, so this shows up as cache pressure rather than as requests.

Local disk got slightly slower on ClickBench. Reading from local disk doesn't use range GETs, so the layout has no effect there. Grouped and row-major 64k files ran within 0.2% of each other on TPC-H. Row-group size still matters on local disk, though. On TPC-H, 64k is about 4% faster than 256k at SF1. On ClickBench's 105-column, string-heavy table it's 3.8% slower (13.01 s vs 12.53 s): four times the morsels means four times the dispatch overhead across a lot of sub-200 ms queries. The best morsel size depends on the workload. We picked 64k knowing that this is the cost to weigh against it.

Two regressions we found on the way

Smaller row groups exposed two problems in the reader. Both are fixed:

• Too many wake-ups. The new block reader woke every waiting worker for each row group it published, where the old pipeline woke one. On a local scan, which publishes one row group at a time, that meant waking all the workers on every submission. Now one item wakes one waiter.
• Plan build grew with row groups × columns. Building the scan plan matched projected column names against every chunk of every kept row group, and created a Python object per chunk. With 1,584 row groups × 105 columns, that took 28.2 ms per query. We now resolve the projected chunks once per file, which takes 7.4 ms.

Together these fixes took local ClickBench from 14.36 s back to 13.01 s on the new layout.

Where this lands

The defaults have changed in Rugo and Opteryx: max_rows_per_row_group is 65,536 and row_groups_per_block is 4. Every path that writes Parquet (INSERT, CREATE TABLE AS, MERGE, OPTIMIZE and the catalog's compaction) streams through the same writer, so new data uses the grouped layout automatically. Existing tables switch over as they're compacted.

If you write Parquet with Rugo yourself, the settings are:

from rugo.parquet import write_parquet

data = write_parquet(morsel, max_rows_per_row_group=65_536, row_groups_per_block=4)

row_groups_per_block=1 gives you a plain row-major file.

There are some things we haven't measured yet:

• Wall time against real GCS. The request counts are real, but they come from a loopback server. The latency saved in production is a modelled estimate (≈130 ms RTT, 32 concurrent GETs), not a measurement.
• The column patcher. ALTER TABLE ADD/DROP COLUMN patches files in place and currently writes them back row-major. They're still correct, and they get the grouping back the next time they're compacted.

The broader lesson

When one setting is doing two jobs, the default is a compromise between them. We first asked "what's the right row-group size?", and there wasn't a right answer, because the size was answering two questions. Once the layout separated how much the engine works on at a time from how much we fetch at a time, each could be set on its own evidence: 64k for the engine, 256k for the network.

Parquet already allowed this. The footer has always pointed at every chunk separately, and putting a row group's chunks together was only a convention. Owning the writer and the reader is what let us change it.

— Justin