Yesterday, we reconstructed a logical plan in a fresh DuckDB connection. That left one gap: the caller still needed its rows back. Today, let's connect that plan to a Flight stream and run our first query across two processes.

I'm keeping the input deliberately boring:

SELECT i, CASE WHEN i = 17 THEN NULL ELSE i END AS nullable
FROM range(5000) t(i);

Both processes can generate this input independently. We can concentrate on the boundary: does the worker return all 5,000 rows, with the right types and NULL, and does the caller learn whether execution actually finished?

One explicit remote call

The native extension now exposes a table function that takes a worker endpoint and SQL text. It plans that SQL inside the coordinator, ships the admitted logical fragment, and presents the returned rows as a DuckDB table. Ordinary queries still run locally.

The order matters:

Preparing or explaining the call performs the local planning work. It doesn't contact the worker. Each execution creates a fresh task identity and deadline, so executing a prepared statement twice doesn't accidentally reuse yesterday's claim.

Compatibility also got stricter. A CMake configuration creates a build ID shared by its shell and extension. Independently configured builds are rejected even when their version strings match. For this first version, I deploy the complete build together. That costs some flexibility, but gives us a concrete boundary for DuckDB's internal plan format.

The supported fragments still use a small set of core operations. I also check casts against the built-in implementation: an extension can replace a cast without changing the source and destination type names. A familiar type name doesn't establish identical behavior.

The stream owns the execution

A connection can't disappear when the request handler returns. DuckDB may still need it to fetch the next batch, and Arrow arrays may still reference buffers from earlier batches.

The stream therefore owns the worker connection, transaction, query result, and registry claim. Each fetch converts a DuckDB chunk through the Arrow C Data Interface into an Arrow record batch. The coordinator converts that batch back into DuckDB vectors. Shared ownership keeps the imported arrays alive for every output column that references them.

The happy path commits the read transaction before reporting completion. Errors and abandoned streams tear down the result and connection before releasing the claim. Cancellation preserves the registry entry while its consumer is still active.

One test exposed a less obvious ordering problem. Flight's C++ client waits for the schema before returning its reader. If the worker starts a long query before sending that schema, the coordinator doesn't yet have a reader it can cancel.

So the worker sends the validated schema first and starts execution when Flight requests the first batch. This also makes an empty result straightforward: it has a schema and a successful end, even though it contains no batches.

Try the process boundary

Here's a standalone transport experiment using ordinary DuckDB and PyArrow Flight. It accepts exactly the SQL above and sends that text in the ticket. The native implementation uses yesterday's registered plan envelope and task ticket; this smaller example lets us inspect the rows without building the extension.

Install uv and save this as remote_select.py in an empty directory. The script declares Python 3.12 and its dependencies so uv can prepare the environment:

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

import sys
import duckdb
import pyarrow as pa
import pyarrow.flight as flight

QUERY = """SELECT i, CASE WHEN i = 17 THEN NULL ELSE i END AS nullable
FROM range(5000) t(i)"""
SCHEMA = pa.schema([("i", pa.int64()), ("nullable", pa.int64())])
ADDRESS = "grpc://127.0.0.1:8815"


class Worker(flight.FlightServerBase):
    def do_get(self, context, ticket):
        if ticket.ticket != QUERY.encode():
            raise ValueError("this demo accepts only its example query")

        def batches():
            with duckdb.connect() as connection:
                reader = connection.execute(QUERY).fetch_record_batch(2048)
                try:
                    yield from reader
                finally:
                    reader.close()

        return flight.GeneratorStream(SCHEMA, batches())


if sys.argv[1:] == ["worker"]:
    with Worker(ADDRESS) as server:
        print("worker listening", flush=True)
        server.serve()
else:
    with flight.connect(ADDRESS) as client:
        options = flight.FlightCallOptions(timeout=5)
        reader = client.do_get(flight.Ticket(QUERY.encode()), options)
        batches = [chunk.data for chunk in reader]
        table = pa.Table.from_batches(batches, schema=reader.schema)
    assert table.schema == SCHEMA
    assert table.column("i").to_pylist() == list(range(5000))
    expected = [None if i == 17 else i for i in range(5000)]
    assert table.column("nullable").to_pylist() == expected
    print(f"{table.num_rows} rows, {len(batches)} batches; values match")

Start the worker in one terminal:

uv run remote_select.py worker

Then run the client in another terminal:

uv run remote_select.py

The client prints 5000 rows, 3 batches; values match. Stop the worker with Ctrl-C. The assertions compare every value, including the NULL; counting rows alone would miss a surprising number of mistakes. This small client collects the result for comparison. The native table function consumes batches incrementally.

Rows aren't proof of success

Suppose the worker sends two batches, then a cast fails in the third. Those first batches have already crossed the network. We can't make them disappear from a consumer that observed them.

The contract is that the query fails. A successful final Flight status is part of the result, and callers must not treat an incomplete prefix as a successful table. I tested this both with a late worker execution error and with a server that deliberately fails after sending a batch.

Cancellation needs similar care. Checking a flag only between batches won't stop a filter that scans for a long time without producing any output. The native worker watches cancellation, disconnects, deadlines, and shutdown, and interrupts the DuckDB connection while execution is active. The caller also cancels a blocked Flight read. An outer LIMIT 1 exercises early cleanup instead of consuming the remaining stream.

For now, calls have bounded deadlines, output batches have a size limit, and each stream has a small cancellation watcher. Blocking network reads are acceptable for this single-worker experiment; scheduling many streams will need a different arrangement.

The native checks now cover scalar values, typed NULLs, empty results, multiple batches, repeated execution, incompatible workers, late errors, and active cancellation. We have the first complete path from a coordinator's logical fragment to rows returned by another process. Next, let's replace the generated range with real input: how much of a Parquet file should one task read?