timsaucer opened a new issue, #25155:
URL: https://github.com/apache/datafusion/issues/25155
### 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<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`:
```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::<T>()` 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).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
