ForeignExecutionPlan cannot be serialized, and host optimizer rules cannot see the plans they are given

[timsaucer]

Describe the bug

A QueryPlanner provided over FFI cannot return a plan containing any node that a host physical optimizer rule inserted, even when that node is a stock DataFusion node with full protobuf support.

Three facts combine:

1. FFI_QueryPlanner returns its plan as protobuf, not as a handle — physical_plan_to_bytes_with_extension_codec on the library side (datafusion/ffi/src/query_planner.rs:168) and physical_plan_from_bytes_with_extension_codec on the host side (:314). So every query through a foreign planner serializes the plan.

2. Physical planning applies session.physical_optimizers(). When the session arrived over FFI those are the host's rules (datafusion/ffi/src/session/mod.rs:765), so each runs back across the boundary and the result returns wrapped in a ForeignExecutionPlan (ForeignPhysicalOptimizerRule::optimize, datafusion/ffi/src/physical_optimizer.rs:312-323). EnsureCooperative is in the default rule list (datafusion/physical-optimizer/src/optimizer.rs:177), so this happens on essentially every plan.

3. ForeignExecutionPlan implements neither try_to_proto (datafusion/physical-plan/src/execution_plan.rs:1025) nor downcast_delegate (:146) — zero matches for either in datafusion/ffi/src/execution_plan.rs. So try_from_physical_plan_with_converter (datafusion/proto/src/physical_plan/mod.rs:1324) skips the native path entirely and falls into the extension-codec arm, which fails at :1382.

The result is that CooperativeExec, which has a perfectly good try_to_proto (datafusion/physical-plan/src/coop.rs:400), becomes unserializable purely by having crossed an FFI boundary.

There is a second, larger consequence. A ForeignExecutionPlan is opaque to downcast_ref, so the host rules that cross the boundary mostly cannot act at all. The stock rules are downcast-driven — enforce_distribution.rs has 20 downcast_ref sites, sort_pushdown.rs 15, window_topn.rs 12 — and against a tree of foreign nodes they match nothing and silently skip. The rules that do fire are the property-driven ones, and those read properties that the FFI bridge fills with wrong defaults (filed separately; see umbrella #25152). So the FFI planner path today gets close to zero real optimization, and the one rule that fires does the wrong thing and then breaks serialization.

To Reproduce

Observed through datafusion-python, whose SessionContext.set_query_planner installs an FFI_QueryPlanner. With a foreign planner installed and no extension codec, planning any query over a Parquet table fails.

Reproduced in-tree by patching the test planner to apply the session's rules — which is what any library planner built on DefaultPhysicalPlanner does — at datafusion/ffi/src/tests/query_planner.rs:92, replacing the bare Ok(Arc::new(EmptyExec::new(schema))):

let mut plan: Arc<dyn ExecutionPlan> = Arc::new(EmptyExec::new(schema));
let config = session.config().options();
for rule in session.physical_optimizers() {
    plan = rule.optimize(plan, config)?;
}
Ok(plan)

Then cargo test -p datafusion-ffi --features integration-tests --test ffi_query_planner test_ffi_query_planner:

Error: Ffi("Internal error: Unsupported plan and extension codec failed with
  [FFI error: This feature is not implemented: PhysicalExtensionCodec is not provided].
  Plan: ForeignExecutionPlan { name: \"CooperativeExec\", ...,
        children: [EmptyExec { ... }] }")

Note the child is concrete — only the host-inserted wrapper is opaque.

Defect (3) alone can be shown with no dylib changes, using the existing AddLimitRule (datafusion/ffi/src/tests/physical_optimizer.rs:31), which inserts a stock GlobalLimitExec across the boundary. Apply that foreign rule to a natively serializable leaf and the result reports name() == "GlobalLimitExec", is not a GlobalLimitExec, and fails physical_plan_to_bytes_with_extension_codec — while the identical plan shape built locally serializes without a codec.

Expected behavior

A foreign planner should be able to return a plan containing stock nodes, and host rules should be able to act on the plans they are handed.

Additional context

The root cause is not "no codec was provided." A codec path for foreign plans does exist: ForeignPhysicalExtensionCodec::try_encode (datafusion/ffi/src/proto/physical_extension_codec.rs:377-383) unwraps the handle back to its home image via FFI_ExecutionPlan::new (datafusion/ffi/src/execution_plan.rs:338) and the peer encodes it. That works for extension nodes. It cannot work for stock nodes, because the bridge only ever reaches the peer's PhysicalExtensionCodec, never the peer's native proto encoder — and no extension codec is contractually expected to encode a built-in node. Supplying DefaultPhysicalExtensionCodec fails identically.

This is the ExecutionPlan-layer counterpart of #22367. That issue documents the same root cause one layer down: FFI_PhysicalExpr carries behavior across the boundary but not identity, so as_any().downcast_ref::<T>() mis-classifies every foreign-wrapped expression. The symptoms differ — #22367 reports silent predicate corruption in the simplifier, this issue reports a hard serialization failure plus inert optimizer rules — but the fix shapes should be considered together.

Two of the three fixes suggested in the original report do not work.

• Report downcast_delegate: not implementable. It returns Option<&dyn ExecutionPlan>; ForeignExecutionPlan holds only an opaque FFI_ExecutionPlan, so there is no local trait object to borrow. Even given one, TypeId is not stable across dylib images, so a cross-image downcast would be unsound.
• Re-attempt the native path after unwrapping the handle: collapses into the try_to_proto option. Unwrapping inside the library yields a plan whose concrete type that image cannot name; the encoding has to happen on the peer side either way.

Directions worth discussing (both need ABI additions; datafusion-ffi gates compatibility on the crate major version at datafusion/ffi/src/lib.rs:64, which bumps each DataFusion major, so they should be batched into one bump):

1. Substitute local instances for built-in rules. Have ForeignSession::physical_optimizers() (datafusion/ffi/src/session/mod.rs:765) return a local instance for each host rule it recognises as built-in, preserving the host's list and order, and wrap only unrecognised rules. Stock rules would then run in the library image on library-local nodes: fully downcast-capable, producing no foreign node, at no serialization cost. Recognition should use an identifier set on FFI_PhysicalOptimizerRule at wrap time, where the host knows the concrete type — matching on rule.name() would silently mistake a customised rule for the stock one, which is the same objection FFI_PhysicalExpr opaque wrapping breaks TypeId downcasts #22367 raises against name()-based dispatch at the plan layer.

   This is structurally the rule-list analogue of the tiered reconstruction proposal in FFI_PhysicalExpr opaque wrapping breaks TypeId downcasts #22367: known built-ins are rebuilt as consumer-local instances, unknown third-party items stay opaque behind the existing vtable. If that model is adopted for PhysicalExpr, applying the same tiering here would be consistent rather than a second, competing mechanism.

2. A try_to_proto backstop for genuinely custom host rules. Add an FFI_ExecutionPlan vtable entry meaning "serialize your subtree with this codec, return bytes", and have ForeignExecutionPlan::try_to_proto call it and prost-decode the result into a PhysicalPlanNode. Recursion terminates because each hop unwraps to a node native to that image.

Considered and set aside, recorded so they are not re-proposed:

• Serialize at the optimizer boundary, per rule or batched at the session. Costs roughly 3× the serialization passes per query, dominated by repeated FileGroup file_groups (datafusion/proto-models/proto/datafusion.proto:1249), so the cost scales with file count rather than plan size. Collapses per-rule error attribution and the observer hook (datafusion/core/src/physical_planner.rs:2963-2980), and loses rule interleaving, which the distributed-planner use case described in the module docs (datafusion/ffi/src/query_planner.rs:20-27) needs. It also cannot deliver the last default rule's output at all until the dynamic-filter converter issue is fixed.
• Serialize any built-in FFI_ExecutionPlan. Requires threading a codec through every FFI_ExecutionPlan::new call site and degenerates into mixed bytes/handle trees.
• Library-owned default rules — the apache/datafusion-python#1721 workaround, which wraps the session so physical_optimizers() returns PhysicalOptimizer::default().rules. It works, but it silently discards any custom host rule, and the in-tree test asserts that rules do cross (datafusion/ffi/src/tests/query_planner.rs:89).

This is not misuse. Rules crossing the boundary is intended behaviour, per that same test. Both three-library tests currently avoid the problem by clearing the rule list (datafusion/ffi/tests/ffi_query_planner.rs:195,266); un-clearing them is a reasonable acceptance criterion for a fix.

Related issues. #22367 (same root cause at the PhysicalExpr layer; tiered-reconstruction proposal). #22329 (FFI_ExecutionPlan missing optimizer-relevant methods — a different set of gaps in the same struct, and a reason foreign nodes are poor optimizer subjects even once they serialize). #24762 and #24106 (codec plumbing at the same planner boundary). #17374 (Stabilize FFI Boundary).

Part of umbrella #25152 covering the FFI planner boundary.

Downstream tracking: apache/datafusion-python#1719 (G1).