You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/08 09:20:50 UTC
(datafusion) 01/01: fix(ffi): prevent recursive session physical planning (#24492)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24492-92746a993c66eb4d97ea044336bff27420800426
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 7401e1bc49aa3aca805b36759f2a391201a0e3b6
Author: Goutam Adwant <86...@users.noreply.github.com>
AuthorDate: Tue Sep 8 09:20:26 2026 +0000
fix(ffi): prevent recursive session physical planning (#24492)
## Which issue does this PR close?
- Closes #24065.
## Rationale for this change
`ForeignSession::create_physical_plan` forwarded through the owning
session. If that session had installed the calling library's query
planner, the callback re-entered the same planner and could recurse
until the stack was exhausted. The callback also returned an
`FFI_ExecutionPlan`, which cannot reconstruct built-in nodes with the
receiving library's Rust type identities for downcasting.
The supported delegation boundary is `FFI_QueryPlanner`: the session
owner exports its original planner before installing a foreign planner,
and the foreign planner retains that handle. This boundary serializes
physical plans and reconstructs them with local type identities.
## What changes are included in this PR?
- Retain the legacy `create_physical_plan` callback at its original
position and signature in `FFI_SessionRef` for DataFusion 55 ABI
compatibility, while making it return an actionable `NotImplemented`
error without invoking the installed planner.
- Make `ForeignSession::create_physical_plan` return the same error
without crossing the FFI boundary or invoking the installed planner.
- Document the captured-`FFI_QueryPlanner` migration in the module
documentation and DataFusion 56 upgrade guide.
- Add unit and cross-library regression coverage for the rejected
direct-delegation path and the successful captured-planner path.
#24690 added the session's logical extension codec to the callback and
fixed planning for logical plans that require custom codec support. It
does not address the two remaining problems covered here: forwarding can
re-enter the installed foreign planner, and the returned
`FFI_ExecutionPlan` does not reconstruct built-in nodes with the
receiving library's Rust type identities. This PR preserves the callback
slot for ABI compatibility but makes direct session planning
unsupported, so its callback-specific codec wiring is no longer
exercised; other FFI codec work remains outside this PR.
## Are these changes tested?
Yes.
- The focused unit regression verifies that both
`ForeignSession::create_physical_plan` and the retained legacy callback
return `NotImplemented` with zero planner re-entry.
- The three-library `dlopen` regression verifies that direct
foreign-session delegation is rejected while the captured
`FFI_QueryPlanner` route succeeds and restores local, downcastable
physical-plan nodes.
- `cargo test -p datafusion-ffi --features integration-tests`
- `cargo clippy -p datafusion-ffi --all-targets --all-features -- -D
warnings`
- `cargo clippy --all-targets --all-features -- -D warnings`
- `RUST_BACKTRACE=1 cargo test --profile ci --exclude
datafusion-examples --exclude datafusion-benchmarks --exclude
datafusion-cli --workspace --lib --tests --bins --features
avro,json,backtrace,extended_tests,recursive_protection,parquet_encryption`
- `RUSTDOCFLAGS="-D warnings" cargo doc -p datafusion-ffi --all-features
--no-deps`
- `cargo fmt --all -- --check`
- `./ci/scripts/doc_prettier_check.sh`
## Are there any user-facing changes?
Yes. `ForeignSession::create_physical_plan` and the retained legacy
callback now return `NotImplemented`; callers must retain and invoke the
session owner's exported `FFI_QueryPlanner`. The `FFI_SessionRef`
callback remains at its original ABI position and signature, so this PR
does not change the struct layout or require consumers to rebuild solely
because of an ABI layout change.
---
datafusion/ffi/src/query_planner.rs | 7 +-
datafusion/ffi/src/session/mod.rs | 159 ++++++++++-----------
datafusion/ffi/src/tests/query_planner.rs | 19 ++-
docs/source/library-user-guide/upgrading/56.0.0.md | 17 +++
4 files changed, 114 insertions(+), 88 deletions(-)
diff --git a/datafusion/ffi/src/query_planner.rs b/datafusion/ffi/src/query_planner.rs
index 6811b91adf..90705a447f 100644
--- a/datafusion/ffi/src/query_planner.rs
+++ b/datafusion/ffi/src/query_planner.rs
@@ -44,9 +44,10 @@
//! C commonly wants A's built-in planning as a starting point, then rewrites the
//! result. A must export its planner *before* installing C's planner on the
//! session, and C must retain that handle: after the swap,
-//! [`Session::query_planner`] reports C's own planner, and
-//! [`Session::create_physical_plan`] dispatches to it, so either one is a
-//! self-call. Delegating to the retained handle is safe, because DataFusion's
+//! [`Session::query_planner`] reports C's own planner, so invoking it is a
+//! self-call. [`crate::session::ForeignSession::create_physical_plan`] is
+//! unsupported because forwarding through A's session would likewise re-enter
+//! C's planner. Delegating to the retained handle is safe, because DataFusion's
//! built-in physical planner never re-dispatches through [`Session`].
//!
//! Retain the planner rather than the session. [`FFI_QueryPlanner`] owns a
diff --git a/datafusion/ffi/src/session/mod.rs b/datafusion/ffi/src/session/mod.rs
index 14ab5a0498..c3a7519cf5 100644
--- a/datafusion/ffi/src/session/mod.rs
+++ b/datafusion/ffi/src/session/mod.rs
@@ -21,10 +21,10 @@
//!
//! Consider a session owned by library A that uses a query planner owned by
//! library C. After A installs C's planner, [`ForeignSession::query_planner`]
-//! returns C's planner and [`ForeignSession::create_physical_plan`] dispatches
-//! to C's planner. C must not call `create_physical_plan`, or invoke the planner
-//! returned by `query_planner`, to delegate planning back to A. Repeating either
-//! self-call recurses until the stack is exhausted.
+//! returns C's planner. Invoking it to delegate planning back to A is a direct
+//! self-call. [`ForeignSession::create_physical_plan`] is deliberately
+//! unsupported because dispatching through A's session would likewise re-enter
+//! C's planner.
//!
//! To delegate safely, A must export its original planner before installing C's
//! planner, and C must retain and invoke that planner directly. See the
@@ -40,7 +40,7 @@ use arrow_schema::ffi::FFI_ArrowSchema;
use async_ffi::{FfiFuture, FutureExt};
use async_trait::async_trait;
use datafusion_common::config::{ConfigFileType, ConfigOptions, TableOptions};
-use datafusion_common::{DFSchema, DataFusionError};
+use datafusion_common::{DFSchema, DataFusionError, not_impl_err};
use datafusion_execution::TaskContext;
use datafusion_execution::config::SessionConfig;
use datafusion_execution::runtime_env::RuntimeEnv;
@@ -117,6 +117,8 @@ pub(crate) struct FFI_SessionRef {
logical_plan_serialized: SVec<u8>,
) -> FFI_Result<SVec<u8>>,
+ /// Retained at its original position for ABI compatibility with consumers
+ /// compiled against DataFusion 55. Direct session planning is unsupported.
create_physical_plan:
unsafe extern "C" fn(
&Self,
@@ -239,31 +241,15 @@ unsafe extern "C" fn optimize_fn_wrapper(
}
unsafe extern "C" fn create_physical_plan_fn_wrapper(
- session: &FFI_SessionRef,
- logical_plan_serialized: SVec<u8>,
+ _session: &FFI_SessionRef,
+ _logical_plan_serialized: SVec<u8>,
) -> FfiFuture<FFI_Result<FFI_ExecutionPlan>> {
- unsafe {
- let runtime = session.runtime().cloned();
- let session = session.clone();
- async move {
- let logical_codec: Arc<dyn LogicalExtensionCodec> =
- (&session.logical_codec).into();
- let session = session.inner();
- let task_ctx = session.task_ctx();
-
- let logical_plan =
- sresult_return!(logical_plan_from_bytes_with_extension_codec(
- logical_plan_serialized.as_slice(),
- task_ctx.as_ref(),
- logical_codec.as_ref(),
- ));
-
- let physical_plan = session.create_physical_plan(&logical_plan).await;
-
- sresult!(physical_plan.map(|plan| FFI_ExecutionPlan::new(plan, runtime)))
- }
- .into_ffi()
+ async move {
+ sresult!(not_impl_err!(
+ "FFI_SessionRef::create_physical_plan is unsupported; export and invoke an FFI_QueryPlanner captured before installing a foreign planner"
+ ))
}
+ .into_ffi()
}
unsafe extern "C" fn create_physical_expr_fn_wrapper(
@@ -529,11 +515,11 @@ impl FFI_SessionRef {
/// # Query planner delegation
///
/// If the session owner installed the current foreign query planner,
-/// [`Session::create_physical_plan`] dispatches back to that planner and
/// [`Session::query_planner`] returns that planner. The planner must retain and
-/// invoke the session owner's previous planner instead of using either method to
-/// delegate back to the session. Otherwise, repeated delegation exhausts the
-/// stack. See [`crate::query_planner`] for details.
+/// invoke the session owner's previous planner rather than delegate back through
+/// the session. [`Session::create_physical_plan`] returns an error because such
+/// delegation would re-enter the installed planner. See [`crate::query_planner`]
+/// for details.
#[derive(Debug)]
pub struct ForeignSession {
session: FFI_SessionRef,
@@ -748,24 +734,11 @@ impl Session for ForeignSession {
async fn create_physical_plan(
&self,
- logical_plan: &LogicalPlan,
+ _logical_plan: &LogicalPlan,
) -> datafusion_common::Result<Arc<dyn ExecutionPlan>> {
- unsafe {
- let codec: Arc<dyn LogicalExtensionCodec> =
- (&self.session.logical_codec).into();
- let logical_plan =
- logical_plan_to_bytes_with_extension_codec(logical_plan, codec.as_ref())?;
- let physical_plan = df_result!(
- (self.session.create_physical_plan)(
- &self.session,
- logical_plan.as_ref().into()
- )
- .await
- )?;
- let physical_plan = <Arc<dyn ExecutionPlan>>::try_from(&physical_plan)?;
-
- Ok(physical_plan)
- }
+ not_impl_err!(
+ "ForeignSession::create_physical_plan is unsupported; export and invoke an FFI_QueryPlanner captured before installing a foreign planner"
+ )
}
fn create_physical_expr(
@@ -859,20 +832,37 @@ mod tests {
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
- use arrow::array::record_batch;
use arrow_schema::{DataType, Field, Schema};
- use datafusion::catalog::{MemTable, MemoryCatalogProvider};
+ use datafusion::catalog::MemoryCatalogProvider;
use datafusion::execution::SessionStateBuilder;
- use datafusion_common::DataFusionError;
+ use datafusion_common::{DataFusionError, Result, exec_err};
use datafusion_expr::col;
use datafusion_expr::registry::FunctionRegistry;
use datafusion_proto::logical_plan::DefaultLogicalExtensionCodec;
use super::*;
- use crate::proto::physical_extension_codec::tests::TestExtensionCodec;
static QUERY_PLANNER_CALLS: AtomicUsize = AtomicUsize::new(0);
static PHYSICAL_OPTIMIZER_CALLS: AtomicUsize = AtomicUsize::new(0);
+ static REENTERING_PLANNER_CALLS: AtomicUsize = AtomicUsize::new(0);
+
+ #[derive(Debug)]
+ struct ReenteringQueryPlanner;
+
+ #[async_trait]
+ impl QueryPlanner for ReenteringQueryPlanner {
+ async fn create_physical_plan(
+ &self,
+ logical_plan: &LogicalPlan,
+ session: &dyn Session,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ if REENTERING_PLANNER_CALLS.fetch_add(1, Ordering::Relaxed) == 0 {
+ session.create_physical_plan(logical_plan).await
+ } else {
+ exec_err!("query planner was re-entered through the session")
+ }
+ }
+ }
unsafe extern "C" fn counting_query_planner(
session: &FFI_SessionRef,
@@ -983,12 +973,6 @@ mod tests {
.await?;
assert_eq!(planned.name(), "EmptyExec");
- let physical_plan = foreign_session.create_physical_plan(&logical_plan).await?;
- assert_eq!(
- format!("{physical_plan:?}"),
- "EmptyExec { schema: Schema { fields: [], metadata: {} }, partitions: 1, cache: PlanProperties { eq_properties: EquivalenceProperties { eq_group: EquivalenceGroup { map: {}, classes: [] }, oeq_class: OrderingEquivalenceClass { orderings: [] }, oeq_cache: OrderingEquivalenceCache { normal_cls: OrderingEquivalenceClass { orderings: [] }, leading_map: {} }, constraints: Constraints { inner: [] }, schema: Schema { fields: [], metadata: {} } }, partitioning: UnknownPartitioning(1) [...]
- );
-
assert_eq!(
format!("{:?}", foreign_session.default_table_options()),
format!("{:?}", state.default_table_options())
@@ -1015,35 +999,48 @@ mod tests {
Ok(())
}
- /// `create_physical_plan` must serialize with the session's logical codec on
- /// both sides of the boundary. A plan that scans a custom table provider is
- /// unserializable without it.
#[tokio::test]
- async fn test_create_physical_plan_uses_logical_codec() -> Result<(), DataFusionError>
- {
- let (ctx, task_ctx_provider) = crate::util::tests::test_session_and_ctx();
-
- let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)]));
- let batch = record_batch!(("a", Int32, [1, 2, 3]))?;
- let table = MemTable::try_new(schema, vec![vec![batch]])?;
- ctx.register_table("test_table", Arc::new(table))?;
+ async fn test_foreign_session_rejects_create_physical_plan() {
+ REENTERING_PLANNER_CALLS.store(0, Ordering::Relaxed);
+ let (ctx, task_ctx_provider) = crate::util::tests::test_session_and_ctx();
+ let state = SessionStateBuilder::new_from_existing(ctx.state())
+ .with_query_planner(Arc::new(ReenteringQueryPlanner))
+ .build();
let logical_codec = FFI_LogicalExtensionCodec::new(
- Arc::new(TestExtensionCodec),
+ Arc::new(DefaultLogicalExtensionCodec {}),
None,
task_ctx_provider,
);
-
- let state = ctx.state();
let local_session = FFI_SessionRef::new(&state, None, logical_codec);
- let foreign_session = ForeignSession::try_from(&local_session)?;
-
- let logical_plan = ctx.table("test_table").await?.into_optimized_plan()?;
- let physical_plan = foreign_session.create_physical_plan(&logical_plan).await?;
-
- assert_eq!(physical_plan.name(), "DataSourceExec");
- assert_eq!(physical_plan.schema().field(0).name(), "a");
+ let foreign_session = ForeignSession::try_from(&local_session).unwrap();
+
+ let error = foreign_session
+ .create_physical_plan(&LogicalPlan::default())
+ .await
+ .unwrap_err();
+
+ assert_eq!(REENTERING_PLANNER_CALLS.load(Ordering::Relaxed), 0);
+ assert!(matches!(error, DataFusionError::NotImplemented(_)));
+ assert!(
+ error
+ .to_string()
+ .contains("export and invoke an FFI_QueryPlanner captured before")
+ );
- Ok(())
+ // An already-compiled DataFusion 55 consumer calls this retained slot
+ // directly. It must receive the same safe failure without re-entering
+ // the installed planner.
+ let callback_error = unsafe {
+ (local_session.create_physical_plan)(&local_session, SVec::new())
+ .await
+ .unwrap_err()
+ };
+ assert_eq!(REENTERING_PLANNER_CALLS.load(Ordering::Relaxed), 0);
+ assert!(
+ callback_error
+ .as_str()
+ .contains("export and invoke an FFI_QueryPlanner captured before")
+ );
}
}
diff --git a/datafusion/ffi/src/tests/query_planner.rs b/datafusion/ffi/src/tests/query_planner.rs
index 40c237c39e..9b5d6a0780 100644
--- a/datafusion/ffi/src/tests/query_planner.rs
+++ b/datafusion/ffi/src/tests/query_planner.rs
@@ -21,7 +21,7 @@ use std::sync::Arc;
use arrow::datatypes::{DataType, Field, Schema};
use async_trait::async_trait;
use datafusion_catalog::default_table_source::source_as_provider;
-use datafusion_common::{Result, exec_err};
+use datafusion_common::{DataFusionError, Result, exec_err};
use datafusion_expr::LogicalPlan;
use datafusion_physical_plan::ExecutionPlan;
use datafusion_physical_plan::empty::EmptyExec;
@@ -115,9 +115,7 @@ impl QueryPlanner for SwappedQueryPlanner {
}
// After the swap, the planner installed on library A's session is this
- // planner, so `session.query_planner()` and `session.create_physical_plan()`
- // are both self-references. Assert the hazard instead of triggering it:
- // calling either would recurse until the stack is exhausted.
+ // planner, so `session.query_planner()` is a self-reference.
let installed = session.query_planner();
let installed: &dyn Any = installed.as_ref();
if installed.downcast_ref::<Self>().is_none() {
@@ -126,6 +124,19 @@ impl QueryPlanner for SwappedQueryPlanner {
);
}
+ // Direct session delegation used to re-enter this planner recursively.
+ // A foreign session must reject it before dispatching to the installed
+ // planner, while the captured planner path below remains usable.
+ let direct_error = session
+ .create_physical_plan(logical_plan)
+ .await
+ .expect_err("direct foreign-session planning should be unsupported");
+ if !matches!(direct_error, DataFusionError::NotImplemented(_)) {
+ return exec_err!(
+ "expected direct foreign-session planning to return NotImplemented; got {direct_error}"
+ );
+ }
+
// Delegate to library A. The result crosses the FFI boundary as
// serialized bytes, so library C receives nodes carrying its own local
// Rust type identities.
diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md
index 2be2df94ef..61940bf726 100644
--- a/docs/source/library-user-guide/upgrading/56.0.0.md
+++ b/docs/source/library-user-guide/upgrading/56.0.0.md
@@ -25,6 +25,23 @@
in this section pertains to features and changes that have already been merged
to the main branch and are awaiting release in this version.
+### `ForeignSession::create_physical_plan` is unsupported
+
+`ForeignSession::create_physical_plan` no longer forwards to the library that
+owns the session. It now returns a `NotImplemented` error because forwarding can
+re-enter an installed foreign planner, and the execution-plan handle returned by
+the old callback cannot restore local Rust type identities for downcasting.
+The original `FFI_SessionRef` callback slot remains in place for ABI compatibility
+with DataFusion 55 consumers, but calling that callback returns the same error.
+
+The session-owning library should instead export its original planner as a
+`datafusion_ffi::query_planner::FFI_QueryPlanner` before installing a foreign
+planner. The foreign planner can retain and invoke that handle to receive a
+serialized physical plan reconstructed with local type identities. See the
+`datafusion_ffi::query_planner` module documentation for the complete delegation
+pattern. `ForeignSession::query_planner`, `optimize`, and `physical_optimizers`
+continue to forward to the owning session across the FFI boundary.
+
### `GroupColumn` now requires `values_preserving`
Custom implementations of the public `GroupColumn` trait must implement
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org