FFI `Session::create_physical_plan` could recurse infinitely

[timsaucer]

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 #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

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

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:

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

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.