timsaucer opened a new issue, #24065: URL: https://github.com/apache/datafusion/issues/24065
### Describe the bug The FFI query planner boundary (`FFI_QueryPlanner`) deliberately serializes both the input logical plan and the resulting physical plan so that each library reconstructs plan nodes with its own local Rust type identities and can downcast them. See the module docs in `datafusion/ffi/src/query_planner.rs`. This is about to land in https://github.com/apache/datafusion/pull/24028 `FFI_SessionRef::create_physical_plan` predates that boundary and was not updated to match. It has three problems, the first of which is a hard failure. ### 1. Unbounded recursion when a foreign query planner is installed Consider the intended deployment: library A (for example `datafusion-python`) owns the session, library B owns a custom table provider, and library C (for example `datafusion-distributed`) owns a query planner. Library A builds a session, library C captures a reference to that session's original query planner, and then A rebuilds the session with C's planner installed. This is a real use case because you could have multiple layers of query planners. If C now calls `Session::create_physical_plan` on the session it receives during planning, the call goes: - `ForeignSession::create_physical_plan` in C (`datafusion/ffi/src/session/mod.rs`) - across FFI into `create_physical_plan_fn_wrapper` in A - `SessionState::create_physical_plan` in A which dispatches to `self.query_planner` - which is C's planner, so control re-enters C Unbounded recursion, and the natural-looking call for "give me library A's physical plan" is precisely the one that blows the stack. The supported route is for C to invoke the query planner handle it captured *before* the swap. That is safe: `DefaultPhysicalPlanner` never re-dispatches through `Session::query_planner` or `Session::create_physical_plan` — its only recursion is into itself (`datafusion/core/src/physical_planner.rs:2791`). But nothing in the API signals that `session.create_physical_plan()` is off limits. Note that `session.query_planner()` is unusable for this purpose too, for a related reason: `query_planner_fn_wrapper` returns whatever planner is installed, and `FFI_QueryPlanner::new_with_ffi_codecs` unwraps a `ForeignQueryPlanner` back to its original handle, so post-swap C receives *itself* and the local-marker bypass in `impl From<&FFI_QueryPlanner>` hands back its own `Arc`. Calling it is a direct self-call. ### 2. The result is returned as `FFI_ExecutionPlan`, so it is opaque ```rust create_physical_plan: unsafe extern "C" fn( &Self, logical_plan_serialized: SVec<u8>, ) -> FfiFuture<FFI_Result<FFI_ExecutionPlan>>, ``` Even if the recursion above were resolved, the returned plan arrives as a `ForeignExecutionPlan` — a trait-object proxy over function pointers (`datafusion/ffi/src/execution_plan.rs:373`). It cannot be downcast to a concrete plan type, even for built-ins such as `RepartitionExec`, `CoalesceBatchesExec`, or `UnionExec`. Separately, each library statically links its own copy of the DataFusion crates, so the `TypeId` for a given plan type differs per library and downcasting could not work across the boundary regardless of the proxy layer. Because essentially every `PhysicalOptimizerRule` is downcast-driven, a rule applied to a plan received this way silently no-ops on the whole tree rather than failing loudly. A distributed planner that needs to find `RepartitionExec` nodes and rewrite them into stages cannot do so. ### 3. The logical plan is serialized without the session's logical extension codec ```rust let logical_plan = sresult_return!(logical_plan_from_bytes( logical_plan_serialized.as_slice(), task_ctx.as_ref(), )); ``` and on the consumer side in `impl Session for ForeignSession`: ```rust let logical_plan = logical_plan_to_bytes(logical_plan)?; ``` Both use the codec-less `logical_plan_to_bytes` / `logical_plan_from_bytes`, even though `FFI_SessionRef` carries a `logical_codec` field that the neighbouring `optimize` and `create_physical_expr` wrappers do use. Any logical plan containing a custom `Extension` node, or a table provider that requires a codec to encode, fails to serialize on this path. ### To Reproduce For problems 1 and 2, using the three-library arrangement above: 1. Library A builds a session with the default query planner. 2. C obtains and retains a handle to A's original query planner. 3. A rebuilds the session with C's `FFI_QueryPlanner` installed. 4. A plans a query, so C's planner runs with an `FFI_SessionRef` for the new session. 5. In C's planner, call `session.create_physical_plan(&logical_plan)` — recursion. Call the captured planner handle instead and the plan comes back as C-local, downcastable nodes, which is the behaviour the session method should have. For problem 3, register a table provider in A whose `LogicalExtensionCodec` is required to encode it (as `LibraryALogicalCodec` does in `datafusion/ffi/tests/ffi_query_planner.rs`), then call `session.create_physical_plan` across the boundary. Serialization fails because the default codec is used instead of the session's. ### Expected behavior `FFI_SessionRef::create_physical_plan` should give the same guarantees as the query planner boundary: - Return the physical plan as serialized bytes so the receiver reconstructs known nodes with its own local Rust type identities and can downcast them. The `physical_codec` already on `FFI_SessionRef` should control reconstruction of extension nodes. - Serialize and deserialize the input logical plan with the session's `logical_codec`, matching `optimize_fn_wrapper` and `create_physical_expr_fn_wrapper`. - Not recurse into the installed query planner. Either the method should plan with the session's default planner, or it should be removed from the FFI surface in favour of an explicit accessor for A's default planner, so that C has a documented way to ask for built-in planning without a self-call. ### Additional context Related design points that came up while reviewing the FFI query planner work, worth resolving alongside this: - **A should export its default planner explicitly rather than have C fetch it via `FFI_SessionRef::query_planner`.** `query_planner_fn_wrapper` bakes in the codecs belonging to the session ref it was fetched through. A handle captured from the pre-swap session keeps serializing with the pre-swap codecs, which goes stale if A registers library B's provider (and its codec) afterward — and C cannot re-fetch post-swap, since that returns C itself. A calling `FFI_QueryPlanner::new(default_planner, .., logical_codec, physical_codec)` with the codecs it intends is the only clean route. - **C must capture the planner, not the session.** `FFI_QueryPlanner` owns an `Arc<dyn QueryPlanner>` in its private data and refcounts through `clone_fn_wrapper`, so it outlives A's original session being dropped. `FFI_SessionRef` does not — it holds `&'a dyn Session` with the lifetime erased into `*mut c_void`. A session ref cached across the swap is a dangling read. - **Custom `ExtensionPlanner` implementations in A remain a recursion hazard.** One that calls `session.create_physical_plan()` will bounce back into C even when C correctly uses its captured planner handle. - **The planner-swap topology is covered by `test_query_planner_swap_round_trips_type_identity`.** That test performs the swap, has library C delegate to the planner it captured beforehand, and asserts type identity is restored in both directions. It also asserts that after the swap `session.query_planner()` reports library C's own planner — documenting the self-call hazard without triggering it. Replacing either serialization step with an `FFI_ExecutionPlan` handoff makes it fail with "library A's plan was opaque to library C". The same test is the natural place to add coverage once `session.create_physical_plan` is fixed. ## Additional Concern Additionally, right now we do not have a hard requirement that a user provides a physical codec, but it seems like this would probably add that as a requirement. We should evaluate if it now become a requirement that *every* producer of a FFI table provider, user defined function, etc will have to provide codecs. If so that will greatly increase the burden on library providers. Right now they only need to write their providers and executors in Rust and make very simple export to python (or another library) via FFI. If they must also provide serialization an deserialization just to use these features, it becomes a much greater burden on the downstream users and we should approach with caution. -- 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]
