apache / apache/datafusion

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

Open
#25,155 0 comments 0 reactions 0 assignees View on GitHub
bug ffi
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

### 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)))`:

```rust
let mut plan: Arc = 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`:

```text
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::()` 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 #22367 raises against `name()`-based dispatch at the plan layer.

This is structurally the rule-list analogue of the **tiered reconstruction** proposal in #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).

Contributor guide

Open the contributing guide

Research direction

Start with datafusion/ffi/src/execution_plan.rs, query_planner.rs, physical_optimizer.rs, and session/mod.rs to trace ForeignExecutionPlan serialization and optimizer crossing. Run the datafusion-ffi query-planner test with physical optimizers enabled, then inspect the related physical optimizer tests. Done means host rules can act on returned plans and stock nodes serialize without requiring an extension codec; the ABI direction still needs maintainer agreement.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
backend-api-design, distributed-systems
Issue type
Bug
Difficulty
5/5
Estimated time
Over a week
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.