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

[timsaucer]

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	Independent
1	ForeignExecutionPlan implements neither try_to_proto nor downcast_delegate, so a foreign-wrapped node cannot serialize	no — see #25155
2	FFI_PlanProperties carries neither scheduling_type nor evaluation_type	yes — see #25153
3	FFI serialization uses DefaultPhysicalProtoConverter, severing shared DynamicFilterPhysicalExpr identity	yes — see #25154
4	Host optimizer rules receive foreign trees and are downcast-blind	no — see #25155

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, and host optimizer rules can actually see the plans they are handed (ForeignExecutionPlan cannot be serialized, and host optimizer rules cannot see the plans they are given #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.

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 ForeignExecutionPlan cannot be serialized, and host optimizer rules cannot see the plans they are given #25155 proposes applying to the optimizer rule list. It also already argues against name()-based dispatch, which ForeignExecutionPlan cannot be serialized, and host optimizer rules cannot see the plans they are given #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).