Yesterday, we turned Parquet row groups into assignments. Now I'd like to execute one: read groups 2 and 0 from the first file, group 0 from the second file, and nothing from a repeated occurrence of the first file.

That sounds like a small change to a file reader. But what happens when the files put their columns in different orders? Or one stores an integer in 32 bits while the query expects 64? And if a filter needs a column that the query doesn't return, who keeps that column around?

These details decide whether distributing a query changes its answer.

I call the new table function duckd_scan. Its input is an explicit list of resolved files, with a list of row-group indexes for each occurrence. Its output is a normal DuckDB stream with a bound schema: the column names and types established while preparing the query.

The assignment has one deliberately strict rule: an empty group list means no work. It still carries a schema. Treating [] as “read everything” would turn an empty task into a duplicate scan. Repeated paths remain separate occurrences, and duplicate group indexes within one occurrence are rejected.

Reuse the reader's understanding of a table

My first temptation was to open each file, select columns, cast them, and apply filters myself. That's already much of a scan implementation. DuckDB's multi-file reader has these responsibilities, and its Parquet reader understands the encoded pages underneath. Ordinary Parquet scans support projection and filter pushdown, including skipping groups when their statistics exclude a match. DuckDB's Parquet documentation describes that behavior.

The useful extension point is smaller: change which group the reader receives next.

The multi-file scheduler already protects work assignment with a lock. Each scanning thread owns a local reader state. The implementation replaces the group-selection step and lets the existing decoder read that group. This is the essential control flow, shown as pseudocode:

under the scheduling lock:
    if this occurrence has no groups left:
        move to the next occurrence
    otherwise:
        give the next assigned group to this thread

outside the lock:
    initialize a Parquet scan for that one group
    produce chunks through DuckDB's existing scan pipeline

A filter on value still needs value, even for SELECT id. DuckDB can read both columns, evaluate the filter, and remove value from the output. If a file's type differs from the bound type, the existing mapping and expression machinery handles the conversion and filtering together. Missing required columns and values that cannot be converted produce errors. With union_by_name=true, missing columns follow that option's NULL-filling behavior.

There is one optimization we must stop borrowing: a whole-file row count cannot answer count(*) for an arbitrary subset. I disable the full-file statistics and count shortcuts for this scan. We can add exact subset statistics later if they become useful.

What crosses the process boundary?

The worker needs the assignment, output schema, and options used when binding it. For example, interpreting Parquet binary data as strings changes the SQL type. A worker's different session default must not silently change that type.

Each file occurrence therefore carries its groups through copying, file pruning, and serialization. The serialized scan also retains Parquet and multi-file options, including binary interpretation, filename and row-number columns, and schema mapping. Deserialization opens the explicit paths without expanding globs again. A wildcard expansion on the worker could associate yesterday's indexes with today's file list.

This still assumes files stay unchanged between planning and reading. Shipping group numbers does not create a storage snapshot. Workers also need filesystem support and access to the same files; a path in a message doesn't grant access.

Let's compare the answers

We can test the assignment semantics with ordinary DuckDB and PyArrow. The experiment below creates two files, reverses the second file's column order, and narrows its id type. It compares explicit group reads with an independent DuckDB query over known physical row ranges.

Install uv, save this as check_scan.py, and run uv run check_scan.py. Its inline metadata pins the Python version and dependencies; uv manages the script environment.

This is a model of the assignment contract. It reads groups with PyArrow and explicitly converts these two known schemas. It does not install the native table function or reproduce DuckDB's general schema mapper. PyArrow exposes the group-reading operation through ParquetFile.

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


def assigned_rows(paths, assignments, schema):
    pieces = []
    for path, groups in zip(paths, assignments, strict=True):
        with pq.ParquetFile(path) as reader:
            if len(set(groups)) != len(groups):
                raise ValueError("duplicate group in one occurrence")
            for group in groups:
                if group < 0 or group >= reader.num_row_groups:
                    raise ValueError("group out of range")
                piece = reader.read_row_group(group)
                pieces.append(piece.select(schema.names).cast(schema))
    return pa.concat_tables(pieces) if pieces else pa.Table.from_batches([], schema)


with TemporaryDirectory() as directory, duckdb.connect() as con:
    a, b = [Path(directory) / name for name in ("a.parquet", "b.parquet")]
    source = pa.table({
        "id": range(5000),
        "value": [None if i % 11 == 0 else i % 7 for i in range(5000)],
    })
    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)
    pq.write_table(other, b, row_group_size=2000)
    paths = [str(a), str(b), str(a)]
    # An independent oracle selects physical row ranges in ordinary DuckDB.
    cases = [
        ("all", [[0, 1, 2], [0], [0, 1, 2]], ["true", "true", "true"], 11200),
        ("subset", [[2, 0], [0], []],
         ["file_row_number < 2000 OR file_row_number >= 4000", "true", "false"], 4200),
        ("empty", [[], [], []], ["false", "false", "false"], 0),
    ]
    for name, groups, predicates, count in cases:
        assigned = assigned_rows(paths, groups, source.schema)
        assert assigned.num_rows == count
        con.register("assigned", assigned)
        branches = [
            f"SELECT id::BIGINT AS id, value::BIGINT AS value "
            f"FROM read_parquet(?, file_row_number=true) WHERE ({predicate})"
            for predicate in predicates
        ]
        con.execute("CREATE OR REPLACE TEMP TABLE expected AS " + " UNION ALL ".join(branches), paths)
        checks = [
            "SELECT * FROM {table} ORDER BY id, value NULLS FIRST",
            "SELECT id FROM {table} WHERE value IS NULL ORDER BY id",
            "SELECT value FROM {table} WHERE id >= 2000 AND value < 4 ORDER BY value",
            "SELECT count(*), count(value), sum(id) FROM {table}",
        ]
        for query in checks:
            actual = con.execute(query.format(table="assigned")).fetchall()
            expected = con.execute(query.format(table="expected")).fetchall()
            assert actual == expected, (name, query)
        print(f"{name}: {count} rows; four comparisons passed")

Expected output:

all: 11200 rows; four comparisons passed
subset: 4200 rows; four comparisons passed
empty: 0 rows; four comparisons passed

The third input occurrence matters: the full assignment contains the first file twice, while the subset gives that occurrence no groups. Comparing sorted lists preserves duplicate rows. A set comparison would hide the very mistake we want to detect.

The filter checks also exercise a column absent from the output, and the aggregate check distinguishes count(*) from count(value) when NULLs are present. For the empty assignment, the declared schema lets the query bind even though no batches were read.

The native implementation gets a separate check against ordinary read_parquet, including nested projections, reordered group assignments, multiple scanning threads, and repeated prepared execution. A fresh database executes serialized scan fragments with different binary-reading defaults, and a separate worker process exercises the Flight path. Those checks are what connect the small experiment to the actual implementation.

We now have a leaf operation that can execute an assigned portion of a Parquet input. The next question is where to place that leaf in a larger query. Which part of a DuckDB plan can move to a worker, and which part must stay with the coordinator?