[Umbrella] FFI planner boundary: foreign-wrapped nodes, lost plan properties, and severed dynamic filters #25152

Describe the bug

Investigating "an FFI QueryPlanner cannot return a plan containing a host-inserted node" turned up four distinct defects at the datafusion-ffi planner/optimizer boundary. One is the originally reported symptom, one is its actual trigger, one is unrelated and pre-existing, and one is a design gap. They are filed separately because two are small and independently fixable, while the other two need a design discussion.

Defect 2 is the trigger for defect 1's most common symptom. FFI_PlanProperties (datafusion/ffi/src/plan_properties.rs:38-66) has no accessor for either field, and reconstruction goes through PlanProperties::new, which defaults to SchedulingType::NonCooperative / EvaluationType::Lazy (datafusion/physical-plan/src/execution_plan.rs:1521-1522). So every node crossing FFI misreports both. EnsureCooperative — the only default rule that is property-driven rather than downcast-driven, and the only consumer of these fields anywhere in datafusion/physical-optimizer/src/ — therefore wraps foreign leaves that are already cooperative. That spurious CooperativeExec is the node that then fails to serialize.

Visible directly in the reproduction below: the ForeignExecutionPlan reports scheduling_type: NonCooperative while the EmptyExec it wraps reports Cooperative.

Defect 3 is unrelated to the rest and breaks any FFI planner today. The last rule in the default list is FilterPushdown::new_post_optimization() (datafusion/physical-optimizer/src/optimizer.rs:181), whose product is shared Arc identity between e.g. HashJoinExec.dynamic_filter.filter (datafusion/physical-plan/src/joins/hash_join/exec.rs:892) and the DataSourceExec it prunes at runtime. DeduplicatingProtoConverter exists to preserve exactly this (datafusion/proto/src/physical_plan/mod.rs:1940-1976), and the FFI paths do not use it.

To Reproduce

Patch the in-tree test planner to apply the session's physical optimizer 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)

cargo test -p datafusion-ffi --features integration-tests --test ffi_query_planner test_ffi_query_planner then fails against a stock SessionContext::default():

REPRO: after host rules, root is CooperativeExec foreign=true
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\",
        ..., scheduling_type: NonCooperative, ... },
        children: [EmptyExec { ..., scheduling_type: Cooperative }] }")

Defect 1 can also be demonstrated 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. Applying that foreign rule to a natively serializable leaf yields a node that 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. A standalone test doing this is straightforward to add.

Expected behavior

• Nodes crossing FFI report their real scheduling and evaluation types (FFI_PlanProperties drops scheduling_type and evaluation_type, so every plan crossing FFI misreports both #25153).
• Shared dynamic filter references survive the planner boundary (FFI serialization uses DefaultPhysicalProtoConverter, severing shared dynamic filter references #25154).
• A foreign planner can return a plan containing stock nodes, host optimizer rules can actually see the plans they are handed, and a rule that wraps its output transparently isn't unwound back to a no-op crossing back out (Physical optimizer rules cross the FFI boundary as handles instead of bytes, so plans are opaque to every downcast-driven rule #25155).

Additional context

Host rules crossing the boundary is intended design, not misuse. The in-tree test errors with "physical optimizers did not cross the FFI boundary" if they don't (datafusion/ffi/src/tests/query_planner.rs:89). Both three-library tests currently sidestep the problem by clearing the rule list (datafusion/ffi/tests/ffi_query_planner.rs:195,266).

Why the existing planner-swap test does not catch this. Enabling default rules on test_query_planner_swap_round_trips_type_identity fails on a shape-sensitive assertion (sort input chain: CooperativeExec [C-local] -> EmptyExec [foreign]), not on serialization. In that topology the rule round-trips A→C→A, and FFI_ExecutionPlan::new unwraps a ForeignExecutionPlan back to its home handle (datafusion/ffi/src/execution_plan.rs:338), so library A's rule receives an A-local plan and A serializes an A-local result. Defect 1 fires only when the rule's home image differs from the image doing the serializing — which is the plain two-library case that datafusion-python hits.

That same unwrap has an independent failure mode in the opposite case: when the plan handed to it is not a bare ForeignExecutionPlan but a transparent wrapper around one (any ExecutionPlan opting into downcast_delegate, e.g. datafusion-tracing's InstrumentedExec). The unwrap check goes through the delegating dyn ExecutionPlan::downcast_ref, so it sees through the wrapper and treats the whole thing as the bare foreign plan — discarding the wrapper. See #25155's Direction B.

Suggested sequencing. #25153 and #25154 first: both are small, independent, and correct regardless of how #25155 resolves. #25153 alone stops the reported failure from firing in the common EnsureCooperative case, though it does not fix the general class. #25155 after its design discussion settles, since the leading candidates need ABI additions.

Relationship to existing issues. None of these four is a duplicate, but three have close neighbours:

• FFI_PhysicalExpr opaque wrapping breaks TypeId downcasts #22367 (FFI_PhysicalExpr opaque wrapping breaks TypeId downcasts) is the same root cause as defects 1 and 4, one layer down at the PhysicalExpr level. Its "tiered reconstruction" proposal — rebuild known built-ins as consumer-local instances, leave third-party types opaque — is the model Physical optimizer rules cross the FFI boundary as handles instead of bytes, so plans are opaque to every downcast-driven rule #25155 proposes applying to the optimizer rule list. It also already argues against name()-based dispatch, which Physical optimizer rules cross the FFI boundary as handles instead of bytes, so plans are opaque to every downcast-driven rule #25155 independently reached. These should be designed together.
• FFI: FFI_ExecutionPlan silently drops producer overrides of optimizer-relevant defaults #22329 (FFI_ExecutionPlan silently drops producer overrides of optimizer-relevant defaults) is the same family as defect 2 but a different struct and a different set of gaps: it lists missing methods on FFI_ExecutionPlan, whereas defect 2 is two missing fields on FFI_PlanProperties. Neither field appears in its list. Note also that two entries in FFI: FFI_ExecutionPlan silently drops producer overrides of optimizer-relevant defaults #22329 have since landed — apply_expressions and partition_statistics are both in the FFI_ExecutionPlan vtable today — so that issue is partially stale.
• proto: serialize and dedupe dynamic filters #20416 / Serialize dynamic filters across network boundaries #20418 (both closed) added the dynamic-filter deduplication machinery that defect 3 shows datafusion-ffi never opted into; [DISCUSSION] Future of Dynamic Filters Sync #21207 carries the design context.

Also adjacent at the same boundary: #24762 and #24106 (codec plumbing for FFI planners), and #17374 (Stabilize FFI Boundary).

Sub-issues: #25153, #25154, #25155.

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