`ExecutionPlan` downcast identity is broken both ways at the FFI boundary: `ForeignExecutionPlan` is opaque to host rules, and `FFI_ExecutionPlan::new`'s foreign-echo shortcut is too transparent for wrapper nodes

[timsaucer]

Describe the bug

Two symptoms, one shared cause: datafusion-ffi has no notion of "FFI-internal identity" distinct from dyn ExecutionPlan's public downcast API (is/downcast_ref, and the downcast_delegate hook added in #22557 so wrapper nodes can redirect them). Both directions of crossing the boundary reuse that public API for a job it wasn't built for, and break in opposite ways.

Direction A — import: a plan crossing into the host becomes permanently opaque.

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. The larger consequence: ForeignExecutionPlan is opaque to downcast_ref, so 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.

This side genuinely cannot be fixed by giving ForeignExecutionPlan a downcast_delegate: it holds only an opaque FFI_ExecutionPlan, so there's no local trait object to point at, and TypeId isn't stable across dylib images, so a cross-image downcast would be unsound regardless.

Direction B — export: a transparent wrapper around an already-foreign plan gets discarded.

FFI_ExecutionPlan::new (datafusion/ffi/src/execution_plan.rs:337) has a shortcut to avoid double-wrapping a plan that is already a bare echo of something that crossed FFI:

pub fn new(mut plan: Arc<dyn ExecutionPlan>, runtime: Option<Handle>) -> Self {
    if let Some(plan) = plan.downcast_ref::<ForeignExecutionPlan>() {
        return plan.plan.clone();
    }
    ...

plan.downcast_ref::<T>() here is the inherent dyn ExecutionPlan::downcast_ref (datafusion/physical-plan/src/execution_plan.rs:1175) — the same public, downcast_delegate-aware helper Direction A's rules call, not Any::downcast_ref:

pub fn downcast_ref<T: ExecutionPlan>(&self) -> Option<&T> {
    match self.downcast_delegate() {
        Some(delegate) => delegate.downcast_ref::<T>(),
        None => (self as &dyn Any).downcast_ref(),
    }
}

downcast_delegate exists so a wrapper can redirect public downcasts to an inner plan while keeping its own type an implementation detail — datafusion-tracing's InstrumentedExec opts in exactly for this (fn downcast_delegate(&self) -> Option<&dyn ExecutionPlan> { Some(self.inner.as_ref()) }), which is the case #22557 was written for.

The shortcut above wants something different: "is plan's own concrete type literally ForeignExecutionPlan, so I can hand back its original FFI struct instead of re-wrapping." Because it goes through the delegating downcast_ref, any wrapper that opts into downcast_delegate and happens to wrap a ForeignExecutionPlan is misclassified as being one — and the shortcut returns the plan from before the wrapper was ever applied, discarding it entirely.

This fires on the plain add_physical_optimizer_rule path, no QueryPlanner required: every physical optimizer rule installed over FFI receives a ForeignExecutionPlan as input (it already crossed the boundary once to reach the rule). A rule that wraps the whole tree in a transparent decorator — datafusion-tracing's instrumentation rule is exactly this — has its wrapper silently dropped as soon as with_new_children/replace_children round-trips a wrapped child back across FFI to repackage it, because that repackaging goes through this same shortcut.

Why both symptoms belong in one issue. They are not independent — they're the two ends of the same rule crossing the boundary. Direction A means a recognized host rule (e.g. EnsureCooperative) can't act on the tree at all. Direction B means an unrecognized rule (any third-party rule, e.g. datafusion-tracing's) that does act on the tree has its output silently unwound back to a no-op. Any design for "host rules cross the FFI boundary and actually do something" has to land both, or a fix for one leaves the other still breaking the same use case: recognized rules do nothing (A), custom rules do nothing either but for the opposite reason (B).

To Reproduce

Direction A. 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 { ... }] }")

Also demonstrable with no dylib changes at all, using the existing AddLimitRule (datafusion/ffi/src/tests/physical_optimizer.rs:31), which inserts a stock GlobalLimitExec across the boundary: 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 fine.

Direction B. Verified against datafusion-ffi 55.1.0 by instrumenting local patched copies of datafusion-ffi, datafusion-physical-plan, and datafusion-tracing with eprintln! at the relevant call sites and running a two-node plan (ProjectionExec over DataSourceExec) through datafusion-tracing's instrument_with_info_spans! rule, installed via datafusion-python's SessionContext.add_physical_optimizer_rule. Direct evidence at the shortcut check itself, single call, two facts printed side by side:

plan.downcast_ref::<ForeignExecutionPlan>().is_some() == true
(plan as &dyn Any).type_id()            == TypeId(0x36d0e1a1b79082b531906aa6608a769e)  // InstrumentedExec
TypeId::of::<ForeignExecutionPlan>()    == TypeId(0x585a73a9c18b03d9190859e857a74dfa)  // different!

Any's real type identity says InstrumentedExec; the delegating downcast_ref used by the shortcut says ForeignExecutionPlan anyway, and the shortcut fires. Consequence: ctx.add_physical_optimizer_rule(<datafusion-tracing's rule, via FFI>) followed by ctx.sql(...).explain() shows the completely unmodified, un-instrumented plan — no InstrumentedExec, no FFI_ExecutionPlan: wrapper, nothing — and the query executes with zero spans created. Wrapping the rule's output in one additional real, non-downcast_delegate-opted-in node (e.g. GlobalLimitExec::new(instrumented, 0, None), a no-op) works around it: explain() then shows FFI_ExecutionPlan: GlobalLimitExec wrapping the (correctly, separately, transparent-by-design) instrumented tree, and spans are created and exported normally.

An in-tree repro for B should be constructable in datafusion-ffi's own test module along the lines of #24722's import_as_foreign! pattern: build a native plan, force it foreign via crate::mock_foreign_marker_id, convert it to a ForeignExecutionPlan via TryFrom<&FFI_ExecutionPlan>, wrap that in a minimal test type whose downcast_delegate returns the inner plan, and assert that FFI_ExecutionPlan::new on the wrapper does not return plan.plan.clone().

Expected behavior

• A foreign planner (or any FFI-crossing rule) can return a plan containing stock nodes, and host rules can act on the plans they are handed (Direction A).
• A transparent wrapper ExecutionPlan — any type implementing downcast_delegate, not just datafusion-tracing's — that wraps a plan re-crossing the datafusion-ffi boundary survives being handed back out, instead of being silently unwound to the plan from before it was applied (Direction B).

Additional context

Not a codec issue (either direction). FFI_ExecutionPlan/FFI_PhysicalOptimizerRule pass live Rust trait objects across FFI via function-pointer vtables (execute, children, with_new_children, ...), not serialized bytes. LogicalExtensionCodec/PhysicalExtensionCodec never enter either path — zero references to either in datafusion/ffi/src/execution_plan.rs.

This is the ExecutionPlan-layer counterpart of #22367, which 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. Its "tiered reconstruction" proposal — rebuild known built-ins as consumer-local instances, leave third-party types opaque — is the model Direction A below proposes applying to the optimizer rule list.

Two of the fixes originally suggested for Direction A do not work:

• Give ForeignExecutionPlan a downcast_delegate: not implementable. It 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. (This is a different proposal from Direction B's fix below — Direction B never touches ForeignExecutionPlan.)
• 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:

1. Substitute local instances for built-in rules (Direction A). 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 needs Direction 3 regardless of how it lands. Unrecognised rules — any third-party rule, including datafusion-tracing's — are explicitly still wrapped and still cross FFI for real under this proposal. That means they still hit Direction B's shortcut bug on the way back out. Direction 3 is a prerequisite for unrecognised rules to work at all under this design, not something tiered reconstruction subsumes.

2. A try_to_proto backstop for genuinely custom host rules (Direction A). 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.

3. Fix the foreign-echo shortcut to use raw Any (Direction B). In FFI_ExecutionPlan::new:

   if let Some(plan) = (plan.as_ref() as &dyn std::any::Any).downcast_ref::<ForeignExecutionPlan>() {
       return plan.plan.clone();
   }

   This is a one-line, backward-compatible fix (it only narrows when the shortcut fires), independent of 1 and 2, and needs no ABI bump — unlike 1 and 2, which need ABI additions and should be batched into one crate-major bump (datafusion/ffi/src/lib.rs:64) if adopted. Worth auditing whether the same delegating-vs-Any confusion exists at other FFI_*::new-style "already foreign" shortcuts while this is fresh — see FFI constructors silently discard arguments when the input is already foreign #24722, the same family of bug (an "already foreign, take the shortcut" fast path misbehaving) for FFI_LogicalExtensionCodec, FFI_PhysicalExtensionCodec, and FFI_TableProvider, though those discard constructor arguments rather than a wrapper node.

Considered and set aside for Direction A, 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 (FFI serialization uses DefaultPhysicalProtoConverter, severing shared dynamic filter references #25154).
• 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, and additionally exercising a transparent third-party wrapper rule (Direction B), is a reasonable acceptance criterion for a fix.

Suggested sequencing. Direction 3 first — it's small, independent of the other two, needs no ABI bump, and is a correctness bug regardless of how A's design lands. Then #25153 and #25154 (small, independent, correct regardless of how Direction A/B resolve). Then Directions 1/2 after design discussion settles, since the leading candidates need ABI additions.

Related issues.

• FFI_PhysicalExpr opaque wrapping breaks TypeId downcasts #22367 (FFI_PhysicalExpr opaque wrapping breaks TypeId downcasts) — same root cause as Direction A, one layer down at the PhysicalExpr level; also already argues against name()-based dispatch, independently reached here for Direction A's rule-recognition design.
• ExecutionPlan downcasting no longer supports transparent wrapper nodes #22557 — added downcast_delegate, precisely for the datafusion-tracing/InstrumentedExec case that Direction B breaks. This issue's Direction B is a consumer of that API misusing it at one call site, not a problem with the API itself.
• FFI: FFI_ExecutionPlan silently drops producer overrides of optimizer-relevant defaults #22329 (FFI_ExecutionPlan silently drops producer overrides of optimizer-relevant defaults) — a different set of gaps in the same struct, and a reason foreign nodes are poor optimizer subjects even once Direction A's serialization is fixed.
• FFI constructors silently discard arguments when the input is already foreign #24722 (closed) — same family as Direction 3: an "already foreign, take the shortcut" fast path in an FFI constructor misbehaving, for FFI_LogicalExtensionCodec/FFI_PhysicalExtensionCodec/FFI_TableProvider rather than FFI_ExecutionPlan.
• proto: serialize and dedupe dynamic filters #20416 / Serialize dynamic filters across network boundaries #20418 (closed) added the dynamic-filter deduplication machinery relevant to FFI serialization uses DefaultPhysicalProtoConverter, severing shared dynamic filter references #25154; [DISCUSSION] Future of Dynamic Filters Sync #21207 carries the design context.
• Also adjacent at the same boundary: FFI_QueryPlanner captures its codecs by value, so a nested foreign planner keeps stale codecs #24762 and Add an FFI extension codec bundle #24106 (codec plumbing for FFI planners), and Stabilize FFI Boundary #17374 (Stabilize FFI Boundary).

Part of umbrella #25152 covering the FFI planner boundary.

Downstream tracking: apache/datafusion-python#1719 (G1), apache/datafusion-python#1739 (Direction B, found via add_physical_optimizer_rule trying to install datafusion-contrib/datafusion-tracing's instrumentation rule from Python).