Physical optimizer rules cross the FFI boundary as handles instead of bytes, so plans are opaque to every downcast-driven rule #25155

Describe the bug

datafusion-ffi has two ways to move a physical plan across the boundary:

• Serialized bytes. FFI_QueryPlanner carries a logical and a physical extension codec (datafusion/ffi/src/query_planner.rs:96-101) and passes plans as protobuf. Nodes reconstruct with the receiving library's Rust type identities, so downcast_ref works on both sides.
• Live handles. FFI_ExecutionPlan passes a function-pointer vtable. Behavior crosses; identity does not. FFI_PhysicalOptimizerRule::optimize takes and returns &FFI_ExecutionPlan and carries no codec at all (datafusion/ffi/src/physical_optimizer.rs:116-140).

The query-planner boundary picked bytes. The optimizer-rule boundary picked handles. Optimizer rules are the one consumer whose entire job is downcast_ref, so they are exactly the consumer that choice breaks.

Every physical optimizer rule installed over FFI receives a ForeignExecutionPlan as input, because the plan already crossed the boundary once to reach the rule. ForeignExecutionPlan holds only an opaque FFI_ExecutionPlan (datafusion/ffi/src/execution_plan.rs:381) and implements neither downcast_delegate (datafusion/physical-plan/src/execution_plan.rs:146) nor try_to_proto (:1025) — zero matches for either in execution_plan.rs. Two consequences:

1. It cannot be serialized. try_from_physical_plan_with_converter (datafusion/proto/src/physical_plan/mod.rs:1324) skips the native path and falls into the extension-codec arm, which fails at :1382. A stock node with a perfectly good try_to_proto becomes unserializable purely by having crossed a boundary.
2. It is invisible to every downcast-driven rule — which is nearly all of them. enforce_distribution.rs has 20 downcast_ref sites, sort_pushdown.rs 15, window_topn.rs 12. Against a tree of foreign nodes they match nothing and silently skip. No error, no warning; the query runs unoptimized.

(2) is the more serious half. (1) at least produces an error message.

This cannot be fixed by giving ForeignExecutionPlan a downcast_delegate: it holds no local trait object to point at, and TypeId is not stable across dylib images, so a cross-image downcast would be unsound regardless.

To Reproduce

No dylib changes needed: the existing AddLimitRule (datafusion/ffi/src/tests/physical_optimizer.rs:31) inserts a stock GlobalLimitExec across the boundary. The result reports name() == "GlobalLimitExec", is not a GlobalLimitExec under downcast_ref, and fails physical_plan_to_bytes_with_extension_codec — while the identical plan shape built locally serializes fine.

Also reproducible 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:

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

Expected behavior

A rule that crosses the FFI boundary receives plan nodes it can downcast, and returns nodes the caller can downcast.

The planner boundary already does this correctly

Worth recording, because it establishes the intended shape and narrows the scope of the fix.

DefaultPhysicalPlanner::create_physical_plan is plan plus optimize in a single call (datafusion/core/src/physical_planner.rs:205), reading its rules off whatever Session it is handed. SessionState::create_physical_plan does not re-run the physical chain afterwards (session_state.rs:828).

When library C delegates to library A's retained planner handle, FFI_SessionRef::new_with_ffi_codecs unwraps C's ForeignSession back to A's original handle (datafusion/ffi/src/session/mod.rs:475), so A's side hits as_local() (:544) and plans with its genuine SessionState. A's rules therefore run in A's image on A-local nodes, and the result reaches C as bytes.

Confirmed by un-clearing the rule list in test_query_planner_swap_round_trips_type_identity and printing what C receives:

SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
  CooperativeExec
    FFI_ExecutionPlan: EmptyExec, number_of_children=0

SortExec and CooperativeExec — the latter inserted by A's EnsureCooperative — both arrive C-local and downcastable. The only foreign node is library B's table-provider scan, a separate axis (#22329). The existing integration tests clear the rule list (datafusion/ffi/tests/ffi_query_planner.rs:195,266), so this property is currently untested.

The failure mode is narrow but easy to hit: if C constructs its own DefaultPhysicalPlanner and hands it the foreign session, session.physical_optimizers() returns ForeignPhysicalOptimizerRules and all 21 rules cross for real. No line of that code looks like optimizer code, which is what makes it a trap.

Measurements

Serialization cost at this boundary has been asserted but never measured. Harness and full results: https://gist.github.com/timsaucer/391695f63b5e1a124a7d92c227a73f5f (drop into datafusion/proto/tests/, run with --release ... --ignored). Measured against main at 55.1.0; ratios are the point, not absolute milliseconds.

Cost model. A physical-plan round trip is linear in file count and schema width: ≈ 0.6 µs and ≈ 55 bytes per (file × column) for a DataSourceExec carrying per-file statistics.

The cost is per-file Statistics, not paths. At 10,000 files × 20 columns:

A real registered parquet listing table populates per-file statistics on 100/100 files and per-file arrow_schema on 0/100, so the second row is the realistic one.

In context, measured end to end on real parquet: at 100 files one round trip of the final plan costs about as much as the entire planning phase (0.37 ms vs 0.36 ms), but stays under 11% of total query time because execution dominates. At the wide-and-partitioned tail that inverts.

Serializing after every rule is not viable — and not because of cost. Running the 21 default rules with a round trip after each, only 11 of 21 crossings succeed. The 10 failures are consecutive, all at encode, all caused by OutputRequirementExec, which OutputRequirements inserts at the top of the chain and OutputRequirements::new_remove_mode() strips near the end:

OutputRequirements, aggregate_statistics, join_selection, LimitedDistinctAggregation,
FilterPushdown, WindowTopN, EnsureRequirements, CombinePartialFinalAggregate,
OptimizeAggregateOrder, ProjectionPushdown

Serializing once per foreign rule is viable. Rules registered via SessionStateBuilder::with_physical_optimizer_rule are appended after the defaults (datafusion/core/src/execution/session_state.rs:1412, :1827), so a foreign rule sees the plan after the marker node is removed. All 12 query shapes tested there encode and decode without error, including TopK and hash-join dynamic filters: plain scan, filter, agg, sort, topk, topk-on-agg, hash join, join+topk, window, union, distinct, subquery. (Encode/decode success is weaker than equivalence for dynamic filters — see Open Questions.)

Leaf Arc identity is stable. 19 of 21 default rules preserve the leaf DataSourceExec by Arc pointer (rebuilt only by FilterPushdown and EnsureRequirements), and the leaf is >95% of plan bytes — so memoizing the encoded leaf is available as a later optimization.

Proposed resolution

The underlying principle, which datafusion-ffi currently applies inconsistently:

Handles for execution, bytes for identity. Anything whose result a consumer must inspect by type — query planners, physical optimizer rules — crosses as serialized bytes so it reconstructs with local type identity. Anything that only needs live behavior — table provider scans, record batch streams — crosses as a handle.

1. Make FFI_PhysicalOptimizerRule carry bytes and a codec, mirroring FFI_QueryPlanner. Replace optimize(&FFI_ExecutionPlan) -> FFI_Result<FFI_ExecutionPlan> with a byte-oriented signature and add an FFI_PhysicalExtensionCodec plus a task-context provider to the struct. The rule then receives library-local nodes and returns library-local nodes; downcast works in both directions. Needs an ABI addition and a crate-major bump (datafusion/ffi/src/lib.rs:64).

   Cost: one round trip per foreign rule, on top of what the caller already pays — sub-millisecond for typical tables per the measurements above. N foreign rules costing N round trips is a usage concern, addressable by composing them into a single rule.

2. Document the planner-delegation rule. A library installing its own QueryPlanner must delegate to the retained handle of the previous planner, and must not hand a foreign session to a locally constructed DefaultPhysicalPlanner. State the mechanism, not just the intent: the standard planner entry point runs session.physical_optimizers() for you.

3. Slim the per-file payload in datafusion-proto. PartitionedFile (datafusion/proto-models/proto/datafusion.proto:1661) carries Statistics per file, ~97% of the round-trip cost, scaling with files × columns. Hoisting or interning it — column statistics are highly redundant across files in a group — recovers most of that. Independently valuable to every datafusion-proto user, and it is what keeps (1) usable at the wide-and-partitioned tail.

Suggested sequencing. (2) and (3) first — independent and correct regardless of how (1) lands. Then #25153 and #25154. Then (1), which needs the ABI bump and reads much better once the payload is slim.

Open questions

• Rules placed mid-chain. with_physical_optimizer_rules(...) lets a caller put a foreign rule anywhere, including inside the span where OutputRequirementExec makes encoding impossible. Do we document "foreign rules run at the end of the chain" as a constraint, or add proto support for that node?

• Custom nodes returned by a foreign rule become a codec requirement. Today a rule returning its own node type works by handle with no codec. Under (1) it needs a registered physical extension codec on both sides or it fails outright. That is a breaking change, and it lands on datafusion-tracing's InstrumentedExec. Arguably an improvement — a loud, actionable error — but it needs a migration note.

• Round-trip equivalence, not just success. The 12 shapes above were verified only to encode and decode without error, which is weaker than it sounds for dynamic filters. A dynamic filter is a shared mutable object: the producing node (SortExec in TopK mode, HashJoinExec) updates it at runtime and the consuming DataSourceExec reads it, so the property that matters is preserved reference identity, not structural similarity. A round trip that hands the scan a distinct-but-equal filter object encodes, decodes, executes, and returns correct results while pruning nothing. See FFI serialization uses DefaultPhysicalProtoConverter, severing shared dynamic filter references #25154 for the converter, proto: serialize and dedupe dynamic filters #20416/Serialize dynamic filters across network boundaries #20418 for the dedup machinery, [DISCUSSION] Future of Dynamic Filters Sync #21207 for design context.

  Two checks would settle it, neither run yet: structurally, assert Arc::ptr_eq between the filters reported by ExecutionPlan::dynamic_expressions_produced (datafusion/physical-plan/src/execution_plan.rs:184) on the producer and those held by the scan after a round trip; behaviorally, compare pruning metrics or rows scanned between the original and round-tripped plans. Comparing result sets will not catch it. This is a correctness prerequisite for (1), since a foreign rule appended at the end of the chain serializes a plan that does contain installed dynamic filters.

• Acceptance criteria. Un-clear the rule lists in datafusion/ffi/tests/ffi_query_planner.rs:195,266, and invert the assertion at datafusion/ffi/src/tests/query_planner.rs:89, which currently asserts rules do cross and so encodes the broken pattern as intended behavior.

Additional context

Not a codec issue in the current code. FFI_ExecutionPlan/FFI_PhysicalOptimizerRule pass live Rust trait objects via function-pointer vtables, not serialized bytes; LogicalExtensionCodec/PhysicalExtensionCodec never enter either path today. Proposal (1) is precisely the change that introduces them there.

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

• Give ForeignExecutionPlan a downcast_delegate — not implementable. No local trait object to borrow, and TypeId is not stable across dylib images.
• Substitute local instances for recognized built-in rules in ForeignSession::physical_optimizers() (datafusion/ffi/src/session/mod.rs:765). Workable, but needs an identifier set at wrap time (matching on rule.name() would silently mistake a customized rule for the stock one — the same objection FFI_PhysicalExpr opaque wrapping breaks TypeId downcasts #22367 raises), carries a permanent registry, and does nothing for third-party rules, which still cross for real. Proposal (1) covers both with no registry.
• try_to_proto backstop on ForeignExecutionPlan via a "serialize your subtree" vtable entry. Fixes serializability but not downcast blindness, so host rules still silently skip.
• Serialize at every rule — not viable, per the measurements: 10 of 21 default rules produce plans containing OutputRequirementExec, which cannot be encoded at all. Also collapses per-rule error attribution and the observer hook (datafusion/core/src/physical_planner.rs:2963-2980), and loses rule interleaving.
• Library-owned default rules — the Add a worked example of three extension libraries in one distributed query datafusion-python#1721 workaround. Works, but silently discards any custom host rule.

This is not misuse. Rules crossing the boundary is intended behavior per the in-tree test.

Related issues.

• FFI_ExecutionPlan::new silently discards transparent wrapper nodes #25717 — FFI_ExecutionPlan's already-foreign fast paths discard transparent wrapper nodes. Related; split from the original report and tracked separately.
• FFI_PhysicalExpr opaque wrapping breaks TypeId downcasts #22367 — FFI_PhysicalExpr opaque wrapping breaks TypeId downcasts; same root cause one layer down, at the PhysicalExpr level.
• FFI: FFI_ExecutionPlan silently drops producer overrides of optimizer-relevant defaults #22329 — FFI_ExecutionPlan silently drops producer overrides of optimizer-relevant defaults; a reason foreign scan nodes remain poor optimizer subjects even after this is fixed.
• FFI_PlanProperties drops scheduling_type and evaluation_type, so every plan crossing FFI misreports both #25153, FFI serialization uses DefaultPhysicalProtoConverter, severing shared dynamic filter references #25154 — small, independent, correct regardless of how this resolves.
• FFI_QueryPlanner captures its codecs by value, so a nested foreign planner keeps stale codecs #24762, Add an FFI extension codec bundle #24106 (codec plumbing for FFI planners), Stabilize FFI Boundary #17374 (Stabilize FFI Boundary).

Part of umbrella #25152 covering the FFI planner boundary.

Downstream tracking: apache/datafusion-python#1719.