Yesterday, we got rows back from a query running in another process. Both sides could generate the same tiny input with range. Now I'd like to give those workers something real to read.

Suppose we have four workers and a directory containing three small Parquet files and one enormous one. Assigning one file to each worker is easy. It also leaves three workers waiting while the fourth finishes the enormous file. Adding more workers won't help if our smallest assignable piece is still that whole file.

Could we cut it into byte ranges? Well... a random byte offset might land in the middle of a compressed page. We need a boundary the reader understands.

Parquet already gives us one: the row group. It's a horizontal slice of a table. Within that slice, each physical leaf column has its own column chunk, and those chunks contain encoded pages. A nested column can have several physical leaves. The row group brings those columns together around the same set of rows. The Parquet format documentation shows how the chunks and footer fit into a file.

That gives us a possible assignment:

Each worker would receive a file location and explicit row-group indexes. The Parquet reader would locate the requested columns inside those groups. The coordinator could later combine several small groups into one task. Today, I want to establish the boundaries before choosing that packing policy.

For this first design, a row group is our smallest assignable input piece. A file written with only one giant row group still offers only one piece. Distribution can't recover parallelism that this choice of boundary doesn't expose. The workers also need access to the same unchanged file; a path on the coordinator's laptop doesn't provide that by itself.

The useful information is near the end of the file. For an ordinary unencrypted Parquet file, the final eight bytes contain a four-byte metadata length followed by PAR1. The reader can locate the serialized footer from that length. The footer describes the schema, row groups, row counts, and column chunk locations. We can inspect it without decoding the rows.

I don't want to maintain another footer parser. The native implementation uses DuckDB's existing Parquet reader through its filesystem, then walks the decoded metadata. For each group, it records the group index, row count, cumulative row offset, and sum of compressed column sizes. The cumulative offset counts logical rows; it isn't a position to seek to in the file.

There's a small trap in the sizes. A row group's total_byte_size describes uncompressed column data, while its aggregate compressed-size field is optional. The individual column metadata has a required compressed-size field that includes page headers. I sum those physical column sizes. The format's metadata definitions spell out that distinction.

Let's inspect an actual file. Install uv and save the following as row_groups.py. It creates its input in a temporary directory and removes it when finished. Python and package versions are declared in the script.

DuckDB's parquet_metadata returns one row per physical column chunk. We group by row-group ID to get one record per work piece. Summing row_group_num_rows would count the same rows once for every physical column, so we take one copy with min and sum only the column sizes.

# /// script
# requires-python = "==3.12.*"
# dependencies = ["duckdb==1.5.4", "pyarrow==19.0.1"]
# ///

from pathlib import Path
from tempfile import TemporaryDirectory

import duckdb
import pyarrow as pa
import pyarrow.parquet as pq

source = pa.table({
    "id": list(range(5000)),
    "value": [None if i % 11 == 0 else i % 7 for i in range(5000)],
})

with TemporaryDirectory() as directory, duckdb.connect() as con:
    path = Path(directory) / "input.parquet"
    pq.write_table(source, path, row_group_size=2000, compression="snappy")
    groups = con.execute("""
        SELECT row_group_id, min(row_group_num_rows),
               sum(total_compressed_size)
        FROM parquet_metadata(?)
        GROUP BY row_group_id
        ORDER BY row_group_id
    """, [str(path)]).fetchall()
    assert [count for _, count, _ in groups] == [2000, 2000, 1000]

    pieces = []
    offset = 0
    with pq.ParquetFile(path) as reader:
        assert len(groups) == reader.num_row_groups
        print("group  first_row  rows  compressed_bytes")
        for index, count, compressed in groups:
            metadata = reader.metadata.row_group(index)
            assert count == metadata.num_rows
            assert compressed == sum(
                metadata.column(c).total_compressed_size
                for c in range(metadata.num_columns)
            )
            print(f"{index:5}  {offset:9}  {count:4}  {compressed:16}")
            piece = reader.read_row_group(index)
            assert piece.num_rows == count
            pieces.append(piece)
            offset += count

    assert offset == source.num_rows
    combined = pa.concat_tables(pieces)
    assert combined.equals(source)
    con.register("by_group", combined)
    expected = con.execute(
        "SELECT * FROM read_parquet(?) ORDER BY id", [str(path)]
    ).fetchall()
    actual = con.execute("SELECT * FROM by_group ORDER BY id").fetchall()
    assert actual == expected
    print(f"{len(groups)} groups, {offset} rows; every value and NULL matches")

Run it with:

uv run row_groups.py

The groups begin at logical rows 0, 2,000, and 4,000, with 2,000, 2,000, and 1,000 rows respectively. The last line reports 3 groups, 5000 rows; every value and NULL matches. Sorting by the unique ID makes the DuckDB comparison independent of scan output order.

The example uses PyArrow's explicit row-group read to exercise each boundary locally. It doesn't send Parquet tasks to yesterday's worker yet. It establishes that the footer gives us pieces we can read separately and reconstruct into the same table. The native helper's checks also cover nested physical columns, empty files, missing column metadata, invalid row counts, and compressed-size overflow.

Those last checks matter before scheduling. Negative sizes must fail instead of turning into enormous unsigned estimates. Group row counts must add up to the file's row count. Reading a valid footer still doesn't prove every data page is intact; execution can discover corruption later, and that must fail the query.

I also wouldn't treat compressed bytes as a cost model yet. A query selecting one small column may read much less than the sum across all columns. Filters can eliminate work. Decompression, decoding, and the operators above the scan cost CPU and memory. Footer and index reads add their own overhead. The size is a useful starting estimate for packing, with limits we can state clearly.

We now have a way to describe pieces inside a file without reading all its rows first. Tomorrow, let's turn those pieces into tasks: how much work belongs in each task, and how do we make sure every input row group appears exactly once?