Shipping a Query to Another Machine: DataFusion's Physical Plan Wire Format and the Expression Deduplication Trick

TL;DR: In a distributed deployment, the coordinator plans the query and the workers execute it, so the plan has to cross a wire. DataFusion encodes physical plans as protobuf. The interesting part is not the encoding but the identity: some expressions are shared between several places in a plan — a dynamic filter used by both a sort and a scan — and if the wire format copied them, they would become two unrelated objects and the sharing would be lost. DataFusion solves this with a stable expression_id and a deserializer that caches by it.

DataFusion is a library. You embed it in a coordinator, you embed it in N workers, and the coordinator sends the plan to the workers. Or you persist plans. Or you send them to a different process for debugging.

Either way, the plan has to become bytes.

datafusion/proto/ is that layer: 2,164 lines in physical_plan/mod.rs alone, plus the generated protobuf messages and a large test suite of round-trip cases.

The extension mechanism

The problem with serialising a plan is that plans contain trait objects. Arc<dyn ExecutionPlan>, Arc<dyn PhysicalExpr>, Arc<dyn ScalarUDF>. There is no way to put a trait object in a protobuf message.

DataFusion’s answer is a codec:

rust
pub trait PhysicalExtensionCodec: Debug + Send + Sync + Any {
    fn try_decode(
        &self,
        buf: &[u8],
        inputs: &[Arc<dyn ExecutionPlan>],
        ctx: &TaskContext,
        proto_converter: &dyn PhysicalProtoConverterExtension,
    ) -> Result<Arc<dyn ExecutionPlan>>;

    fn try_encode(
        &self,
        node: Arc<dyn ExecutionPlan>,
        buf: &mut Vec<u8>,
        proto_converter: &dyn PhysicalProtoConverterExtension,
    ) -> Result<()>;
    ...
}

The built-in operators are handled by the generated code; the extension operators — a custom ExecutionPlan from a third-party crate — are handled by whoever registered them. The codec gets the opaque bytes and the already-decoded children, and reconstructs the node.

Note that the codec also has separate hooks for UDFs:

rust
fn try_decode_udf(&self, name: &str, _buf: &[u8]) -> Result<Arc<dyn ScalarUDF>> {
    not_impl_err!("PhysicalExtensionCodec is not provided for scalar function {name}")
}

and for higher-order functions, aggregate functions, and expressions themselves. Each has a sensible default: decoding fails loudly rather than silently producing a wrong plan.

The identity problem

Now the interesting part.

Some expressions in a physical plan are shared. The clearest example is DynamicFilterPhysicalExpr — the runtime filter we looked at in an earlier article. When a TopK pushes a filter down into a SortExec, and that filter is also referenced by a DataSourceExec several levels down, the plan contains the same logical object in two places.

In memory, that is fine: two Arc pointers to one allocation. Change the filter and both places see it.

On the wire, it is not. Protobuf has no reference type. The encoder writes the expression twice, and the decoder produces two independent objects. Now the TopK updates its copy of the filter, and the scan is reading the other copy. The dynamic filter silently stops working.

The fix is a stable identity on the expression:

rust
/// This id must be preserved across [`PhysicalExpr::with_new_children`] or any other
/// methods which may want to preserve identity.
///
/// Default is `None`: the expression has no identity worth preserving across a
/// serialization boundary.
fn expression_id(&self) -> Option<u64> {
    None
}

Note the default. Most expressions have no identity worth preserving, and saying so explicitly means the mechanism costs nothing for the common case.

DynamicFilterPhysicalExpr overrides it — and it already had to, because of the Hash/Eq contract issue we saw earlier:

rust
/// A unique identifier for the expression.
expression_id: u64,

preserved across update():

rust
*current = Inner {
    // Preserve the expression id across updates.
    expression_id: current.expression_id,
    generation: new_generation,
    expr: new_expr,
    is_complete: current.is_complete,
};

The deduplicating deserializer

The deserializer is a RefCell<HashMap<u64, Arc<dyn PhysicalExpr>>>:

rust
/// Internal deserializer that caches expressions by their `expression_id()` so
/// multiple occurrences of the same expression are deduped.
#[derive(Default)]
struct DeduplicatingDeserializer {
    /// Cache mapping expression_id to deserialized expressions.
    cache: RefCell<HashMap<u64, Arc<dyn PhysicalExpr>>>,
}

and the lookup:

rust
fn proto_to_physical_expr(
    &self,
    proto: &protobuf::PhysicalExprNode,
    input_schema: &Schema,
    ctx: &PhysicalPlanDecodeContext<'_>,
) -> Result<Arc<dyn PhysicalExpr>> {
    // `expr_id` is the generic identity slot on `PhysicalExprNode`.
    // The default serializer populates it from `PhysicalExpr::expression_id`.
    // A missing id means this expression type doesn't participate in deduping.
    let Some(id) = proto.expr_id else {
        return parse_physical_expr_with_converter(proto, input_schema, ctx, self);
    };

    let parsed = parse_physical_expr_with_converter(proto, input_schema, ctx, self)?;

    let mut cache = self.cache.borrow_mut();
    if let Some(cached) = cache.get(&id) {
        // Since expressions may manage their own internal state when deriving
        // expressions via `with_new_children`, we use `with_new_children`
        // to opt into the same behavior.
        //
        // For example, one `DynamicFilterPhysicalExpr` may be derived from
        // another resulting in shared references. Using `with_new_children`
        // is meant to preserve those references.
        let children: Vec<_> = parsed.children().into_iter().cloned().collect();
        return Arc::clone(cached).with_new_children(children);
    }

    cache.insert(id, Arc::clone(&parsed));
    Ok(parsed)
}

Three details:

A missing id means “not participating”. An expression that does not override expression_id is decoded normally, with no cache lookup. The fast path is the default path.

The cached expression is derived, not returned directly. This is the subtle part. The same shared expression can appear in two places with different children — because a projection pushdown rewrote one of them but not the other. Returning the cached object verbatim would give the second place the first place’s children, which is wrong.

So the code takes the cached expression and calls with_new_children with the freshly parsed expression’s children. The identity (the Arc inside the DynamicFilterPhysicalExpr) is preserved; the structural position is correct.

The comment says exactly this:

Since expressions may manage their own internal state when deriving expressions via with_new_children, we use with_new_children to opt into the same behavior.
For example, one DynamicFilterPhysicalExpr may be derived from another resulting in shared references. Using with_new_children is meant to preserve those references.

The cache is per-deserialisation. DeduplicatingProtoConverter creates a fresh DeduplicatingDeserializer for each top-level call:

rust
pub struct DeduplicatingProtoConverter {}

impl PhysicalProtoConverterExtension for DeduplicatingProtoConverter {
    fn proto_to_execution_plan(
        &self,
        proto: &protobuf::PhysicalPlanNode,
        ctx: &PhysicalPlanDecodeContext<'_>,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        let deserializer = DeduplicatingDeserializer::default();
        proto.try_into_physical_plan_with_context(ctx, &deserializer)
    }
    ...

which is correct: the cache exists to share within one plan, not across plans. Two different plans must not accidentally share an expression.

The `RefCell` and the borrow

The cache is a RefCell<HashMap<...>> and the borrow is taken after the parse:

rust
let parsed = parse_physical_expr_with_converter(proto, input_schema, ctx, self)?;

let mut cache = self.cache.borrow_mut();

The ordering matters. parse_physical_expr_with_converter recurses — it may decode children, which may recursively call back into proto_to_physical_expr, which would try to borrow_mut the same RefCell again. Taking the borrow after the parse avoids the re-entrant borrow panic.

That is a subtle and easily-broken ordering, and it is exactly the kind of thing that a comment should explain. It does not have one, which is either a gap or a deliberate “this is obvious if you know RefCell” choice.

The naming problem

There is a second, newer path for custom expressions, and the doc comment contrasts it with the codec approach:

Expressions can instead name themselves on the wire and be decoded by a per-type decoder registered on the session — see ExtensionExprFromProto and register_physical_expr. That path resolves by name rather than by codec registration order, so two crates claiming the same name collide at registration. This method stays the fallback for everything unnamed or unregistered.

So there are two mechanisms, with a documented trade-off: the codec path is order-dependent (registration order decides who handles what), the registry path is name-dependent (and names can collide). The codec path stays for backwards compatibility and as the fallback.

The catch

The deduplication is opt-in per expression type. An expression that does not override expression_id will be copied on the wire, and any shared state it holds will be split. For most expressions that is fine; for a stateful one it is a silent correctness bug. The trait’s doc says “the expression has no identity worth preserving” — but that is a statement the expression author has to get right.
with_new_children must preserve the identity. The dedup logic depends on it. The DynamicFilterPhysicalExpr implementation goes out of its way to do so, and the comment explains why — but it is a convention, not a type-level guarantee.
The RefCell borrow ordering is load-bearing. Reordering those two statements turns a working plan into a panic.
The wire format is a compatibility surface. A proto field renumbering breaks every deployed worker. The round-trip test suite exists for this reason.

Why it matters

The expression-deduplication trick is a small thing — about forty lines — but it is the difference between a dynamic filter that works in a distributed deployment and one that silently does nothing. The failure mode is invisible: the query returns the right answer, just slower, and nothing complains.

That is the shape of a lot of good distributed-systems engineering: the hard part is not the data movement, it is preserving the identity of the things that reference each other.