Yesterday, we worked out what to compare before sending a plan to a worker. Today, let's actually send a message and get something back.

There are two different conversations happening here. One is about the work: which build are you running, can you accept this task, and can you stop it? The other carries the rows produced by that work.

Those responsibilities are usually called the control plane and the data plane. They can share one server and one port. We don't need to deploy two services just because we've given the conversations different names.

What Flight gives us

Arrow Flight provides RPC methods for exchanging metadata, application-defined actions, and streams of Arrow record batches. For our first worker, two methods are enough to explain the path: DoAction carries control messages, and DoGet opens a data stream using a ticket. Flight also has upload and bidirectional exchange methods. Its protocol documentation describes the available request patterns.

The names and meanings of our actions are our responsibility. Flight doesn't know what a DuckDB task is, or whether a second request should rerun it.

Here's the intended conversation once task execution is available:

If the query is abandoned, a cancellation action needs to reach the worker too. The diagram describes our protocol design; Flight supplies the calls that carry it.

Registering work and running it

Why separate task registration from fetching the result? I want the coordinator to finish describing and validating the work before execution starts. Registration can associate a task identifier with a plan fragment, input assignments, and a deadline. Opening the result stream can then claim that registered task.

The ticket only needs enough information to identify the task. Sending the whole plan again on every fetch would make it harder to tell whether we're fetching existing work or asking for new work.

But separating the calls introduces state. What happens if registration succeeds and the client disappears before DoGet? We now have an abandoned task to clean up. What if two clients fetch the same ticket? We'll need an atomic claim if our policy allows only one execution.

So, well... Flight saves us from implementing the transport, but the task lifecycle is still ours. We'll build the envelope and registry next, before allowing arbitrary fragments to execute.

A hello is already useful

For the first C++ worker, I started with a dedicated process listening on loopback and a duckd_hello action. It reports the control protocol version, DuckDB version and source revision, and distributed extension version. A separate probe compares those fields with its own process and reports whether they match.

That is the first compatibility check we can exercise over the network. The richer build manifest from yesterday remains work to add before relying on this for plan compatibility; matching the reported versions alone doesn't prove identical binaries.

I also gave the probe a two-second deadline. A process that never replies should produce an error we can act on. Reading local settings or explaining a probe should remain possible without contacting the worker.

There are three useful outcomes to preserve: a valid matching reply, a valid mismatching reply, and a failed exchange. A mismatching version should appear in the diagnostic result. A malformed message or connection failure should produce an error. Collapsing all three into “worker unavailable” would throw away information we need to fix the problem.

Try both conversations

Let's use a small Python service to see the control and data calls together. It returns two fixed rows so we can focus on the transport; these rows aren't the result of executing a remote SQL query.

With Python 3.12 and a platform supported by PyArrow, create an environment:

python3.12 -m venv .venv
.venv/bin/python -m pip install pyarrow==19.0.1

Save the following as flight_demo.py, then run .venv/bin/python flight_demo.py. The server binds an available loopback port and shuts down when the block exits. PyArrow starts serving when the server object is created.

import pyarrow as pa
import pyarrow.flight as flight


class Worker(flight.FlightServerBase):
    def do_action(self, context, action):
        if action.type != "hello" or action.body.size != 0:
            raise ValueError("expected hello with an empty body")
        yield flight.Result(b"demo-v1")

    def do_get(self, context, ticket):
        if ticket.ticket != b"sample":
            raise KeyError("unknown ticket")
        return flight.RecordBatchStream(
            pa.table({"customer_id": [1, 2], "orders": [1, 2]})
        )


with Worker(("127.0.0.1", 0)) as worker:
    with flight.connect(("127.0.0.1", worker.port)) as client:
        options = flight.FlightCallOptions(timeout=2)
        replies = list(client.do_action(flight.Action("hello", b""), options))
        if len(replies) != 1 or replies[0].body.to_pybytes() != b"demo-v1":
            raise RuntimeError("unexpected worker identity")

        reader = client.do_get(flight.Ticket(b"sample"), options)
        batches = [chunk.data for chunk in reader if chunk.data is not None]
        result = pa.Table.from_batches(batches, schema=reader.schema).to_pydict()
        assert result == {"customer_id": [1, 2], "orders": [1, 2]}
        print("hello: demo-v1")
        print(result)

The output is:

hello: demo-v1
{'customer_id': [1, 2], 'orders': [1, 2]}

The service assigns meaning to hello and sample; neither is a built-in Flight command. The client consumes the action's response stream, checks the identity, and then consumes the record-batch stream to its end.

This example collects the whole result because it contains two rows. In the query engine, we'd feed batches into DuckDB as they arrive. Keeping all batches in a Python list would defeat the memory behavior we're trying to build.

Receiving rows isn't the same as finishing

A stream can return a batch and then fail. If that happens, those rows don't make the query successful. The client must observe how the stream ends, and the coordinator must require every producer it depends on to finish successfully.

The same point applies to control calls. Getting an acknowledgement message is only part of the exchange; the call can still end with an error. That's why our probe expects exactly one identity message and checks for a successful end after it.

I ran the Python example and also exercised the native worker from a separate process: matching and mismatching replies, malformed responses, an unreachable endpoint, a delayed reply, and shutdown. These checks establish the first worker/control path. The native worker still rejects result requests because fragment execution isn't installed yet.

We now have a process we can contact, inspect, and stop, plus a concrete view of how result batches will travel. Tomorrow, let's give registration something meaningful to carry: a versioned envelope containing the plan fragment and the identity of the task that owns it.