Yesterday, we found the row groups inside a Parquet file and read them separately. That gives us boundaries a worker can understand. But how many groups should we put in one task?

One group per task sounds reasonable until we encounter thousands of tiny groups. Each task still needs registration, execution, and a result stream. At the other extreme, putting everything into one task gives us the same waiting problem we had with whole files.

I'd like a byte target: keep adding consecutive row groups while their estimated compressed size fits. When the next group would exceed the target, start another split. A split is an assignment of input groups; a later scheduling step will choose the worker that executes it.

Here's a schematic example with a 100 MiB target:

The 130 MiB group gets a split of its own. Our reader boundary is still a complete row group, so the target cannot be a hard maximum. A smaller target won't divide that group. DuckDB's Parquet guidance describes the related tradeoff between group size, parallelism, and per-group overhead.

For now, I pack within each file occurrence and keep the original group order. This leaves small files as separate splits. We can revisit combining them when task overhead becomes a measured problem. Sorting all groups by size might fill splits more tightly, but it would also rearrange the reads. We don't need that policy to establish correct assignments.

Compressed bytes remain an estimate of input work. Projection may read only a few columns, filters may skip groups, and decoding has its own cost. The target controls granularity; it doesn't promise equal running times or bounded memory.

There is also an identity detail that's easy to lose. Consider the explicit input list [a.parquet, b.parquet, a.parquet]. The last entry is another occurrence of the first file. Deduplicating paths would remove rows from the query. I identify a group by (input occurrence, group index), so occurrence 0, group 2 and occurrence 2, group 2 are separate pieces of work.

The native planner now opens files through DuckDB's existing task scheduler. Each task writes its splits into a slot belonging to that input occurrence. After all tasks finish, we concatenate those slots in input order. Completion order therefore doesn't affect assignment order. With one configured thread, the same path works sequentially. An error fails planning after scheduled work has drained, while the referenced file list and results are still alive.

Let's try the packing rule with real files, including a repeated path. Install uv and save this as pack_row_groups.py. It creates and removes its own temporary inputs. The small Python experiment uses a standard thread pool for footer reads; the native implementation uses DuckDB's scheduler.

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

from collections import Counter
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from tempfile import TemporaryDirectory

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


def sizes(path):
    with pq.ParquetFile(path) as reader:
        result = []
        for index in range(reader.num_row_groups):
            group = reader.metadata.row_group(index)
            result.append(sum(group.column(c).total_compressed_size
                              for c in range(group.num_columns)))
        return result


def pack(group_sizes, target):
    if target <= 0:
        raise ValueError("target must be positive")
    splits, indexes, used = [], [], 0
    for index, size in enumerate(group_sizes):
        if size < 0:
            raise ValueError("negative compressed size")
        if indexes and (used > target or size > target - used):
            splits.append((indexes, used))
            indexes, used = [], 0
        indexes.append(index)
        used += size
    if indexes:
        splits.append((indexes, used))
    return splits


assert pack([40, 60, 130, 0], 100) == [([0, 1], 100), ([2], 130), ([3], 0)]

with TemporaryDirectory() as directory, duckdb.connect() as con:
    a, b = (Path(directory) / name for name in ("a.parquet", "b.parquet"))
    source = pa.table({
        "id": list(range(5000)),
        "value": [None if i % 11 == 0 else i % 7 for i in range(5000)],
    })
    # Same names, reversed physical order, and a narrower integer in b.
    other = pa.table({
        "value": [None if i % 11 == 0 else i % 7 for i in range(5000, 6200)],
        "id": pa.array(range(5000, 6200), type=pa.int32()),
    })
    pq.write_table(source, a, row_group_size=2000, compression="snappy")
    pq.write_table(other, b, row_group_size=400, compression="snappy")
    paths = [str(a), str(b), str(a)]
    with ThreadPoolExecutor(max_workers=4) as pool:
        metadata = list(pool.map(sizes, paths))
    target = sum(metadata[0][:2])
    splits = [
        (occurrence, indexes, used)
        for occurrence, group_sizes in enumerate(metadata)
        for indexes, used in pack(group_sizes, target)
    ]

    expected_groups = Counter(
        (occurrence, index)
        for occurrence, group_sizes in enumerate(metadata)
        for index in range(len(group_sizes))
    )
    assigned_groups = Counter(
        (occurrence, index)
        for occurrence, indexes, _ in splits
        for index in indexes
    )
    assert assigned_groups == expected_groups

    pieces = []
    for occurrence, indexes, used in splits:
        assert used <= target or len(indexes) == 1
        with pq.ParquetFile(paths[occurrence]) as reader:
            piece = reader.read_row_groups(indexes)
        pieces.append(piece.select(source.schema.names).cast(source.schema))
        print(f"input {occurrence}, groups {indexes}, {used} compressed bytes")

    con.register("assigned", pa.concat_tables(pieces))
    expected = con.execute("""
        SELECT * FROM read_parquet(?, union_by_name=true)
        ORDER BY id, value NULLS FIRST
    """, [paths]).fetchall()
    actual = con.execute(
        "SELECT * FROM assigned ORDER BY id, value NULLS FIRST"
    ).fetchall()
    assert actual == expected
    assert len(actual) == 11200
    print(f"{len(splits)} splits, {len(assigned_groups)} group occurrences, "
          f"{len(actual)} rows; values and NULLs match")

Run it with:

uv run pack_row_groups.py

The result is five splits covering nine group occurrences and 11,200 rows. The coverage assertion uses a counter, so it detects both missing groups and repeated assignments. Comparing the reconstructed rows with ordinary DuckDB also checks that the repeated path contributes its rows twice.

Notice the column selection and cast before concatenation. File b stores the columns in reverse order and uses a narrower integer for id. The example explicitly maps names and widens that integer to our known output schema. DuckDB's baseline uses union_by_name to reconcile these files. That option can also introduce NULLs for missing columns, which changes the scan's contract.

A split containing only a path and group indexes cannot express all of that. The execution plan also needs the bound output schema, column mapping, and the options that determine interpretation, such as schema union and partition-column handling. Workers must not infer those independently from whichever file they happen to receive first. Today's native packer assigns groups using the resolved input list; implementing the scan contract is tomorrow's work. The example handles these two known schemas with an explicit conversion.

The native checks now cover exact targets, oversized and zero-byte groups, integer limits, empty inputs, repeated paths, and overlapping footer opens. In C++, I check the remaining capacity before adding sizes so the arithmetic itself cannot overflow. Files still need to remain unchanged between inspection and execution.

We have a deterministic assignment with every input group represented once, including repeated file occurrences. Next, let's make DuckDB execute those explicit assignments while preserving the behavior of the original scan.