Skip to content

Commit 282e3a5

Browse files
timsaucerclaude
andcommitted
feat: rebind foreign codecs when a planner install forks the session
Installing a foreign query planner writes to `SessionState`, and `with_query_planner` must not modify its receiver, so it forks. A foreign codec holds an `FFI_TaskContextProvider` pointing at the session it was installed on, and until now the fork could not move it: passing a new provider to `FFI_LogicalExtensionCodec::new` was silently discarded whenever the codec was already foreign. The fork rebound only its own outer wrapper, so decode callbacks in the extension library kept answering from the pre-fork registry, and the pre-fork session had to be retained or the weakly held provider dangled. apache/datafusion#24722 fixes the discard; those constructors now adopt the provider on the already-foreign path. Repoint the patch at the branch carrying it and rebind both codecs onto the fork. Verified the branch carries everything already pinned rather than trusting the commit graph, which reports the two as diverged: across 3811 files the only differences are the four constructors from the fix, and `datafusion/ffi/src/session/mod.rs` is byte-identical, so the `create_physical_plan` codec fix arrives as its branch-55 backport. `ancestors` and its helpers are deleted. They existed only to keep the pre-fork session alive for a codec that could not be moved off it, and a codec bound to the running session needs no such anchor. Three tests, replacing two that were weaker than they looked. One registers a function on the fork after the codec was installed on its parent and resolves it, which is the direct evidence the rebind happened; it failed before this change. One installs a planner twice and asserts the first context still cannot resolve a function registered only on the second, covering the clone-before-adopt half — a rebind that mutated the shared handle would pass the first test and fail this one. The third keeps the live-handle case. The test it replaces required a name registered nowhere, so it passed for the same reason as the negative control and never exercised a fork at all. Note the version floor in `Cargo.toml` rather than raising it now: the patched branch still reports 55.0.0, so the requirement can only move to 55.1.0 when the patch section is removed. Building against 55.0.0 without the patch would compile and silently skip the rebind. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1 parent 9696956 commit 282e3a5

5 files changed

Lines changed: 161 additions & 97 deletions

File tree

‎.ai/skills/ffi-capsule-protocol/SKILL.md‎

Lines changed: 14 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -116,14 +116,22 @@ one of them, and the failure is a bare `TypeError` from a `call1`. So:
116116
`python/datafusion/user_defined.py`, where the `Protocol` type hints for
117117
these methods live.
118118

119-
## Rule 6 — the session a codec captures is not always the running session
119+
## Rule 6 — a fork must rebind the codecs it carries
120120

121121
Installing a foreign query planner **forks** the session
122-
(`PySessionContext::ctx_with_rebound_planner`). A codec installed beforehand
123-
keeps pointing at the pre-fork session, so registrations made after the fork
124-
are invisible to it. Install codecs before the planner, which is what the
125-
contributor guide already advises. `PySessionContext::ancestors` keeps forked
126-
parents alive so the captured session cannot be dropped out from under a codec.
122+
(`PySessionContext::derived_parts`), because installing one writes to
123+
`SessionState` and the receiver must not be modified. A foreign codec holds an
124+
`FFI_TaskContextProvider` pointing at the session it was installed on, so the
125+
fork rebinds each one onto itself via
126+
`PySessionContext::rebound_{logical,physical}_codec`. Skip that and decode
127+
callbacks answer from the pre-fork registry.
128+
129+
Rebinding relies on `FFI_{Logical,Physical}ExtensionCodec::new` adopting the
130+
provider on the already-foreign path, which needs DataFusion 55.1.0 or newer
131+
(apache/datafusion#24722). It clones the handle before overwriting, so the
132+
receiver keeps its own binding — assert that, not just that the fork works.
133+
A codec this library owns round-trips unchanged, so the rebind is safe to apply
134+
unconditionally.
127135

128136
## Where the truth is
129137

‎Cargo.toml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,11 @@ codegen-units = 2
7777

7878
# We cannot publish to crates.io with any patches in the below section. Developers
7979
# must remove any entries in this section before creating a release candidate.
80+
#
81+
# When these are removed, raise the DataFusion requirement above to 55.1.0. The
82+
# FFI codec rebinding in `PySessionContext::derived_parts` is a silent no-op
83+
# before that release (apache/datafusion#24722), so a 55.0.0 build would compile
84+
# and then resolve decode callbacks against the pre-fork session.
8085
[patch.crates-io]
8186
datafusion = { git = "https://github.com/timsaucer/datafusion", branch = "fix/ffi-constructor-argument-drop-55" }
8287
datafusion-substrait = { git = "https://github.com/timsaucer/datafusion", branch = "fix/ffi-constructor-argument-drop-55" }

‎crates/core/src/context.rs‎

Lines changed: 77 additions & 63 deletions
Original file line numberDiff line numberDiff line change
@@ -373,16 +373,6 @@ pub struct PySessionContext {
373373
pub ctx: Arc<SessionContext>,
374374
logical_codec: Arc<PythonLogicalCodec>,
375375
physical_codec: Arc<PythonPhysicalCodec>,
376-
/// Sessions this one was derived from, held only to keep them alive.
377-
///
378-
/// A foreign codec installed here is built against the session current at
379-
/// the time of the call, and `FFI_TaskContextProvider` holds that session
380-
/// weakly. Installing a foreign query planner afterwards forks, and
381-
/// rebinding the Python name would otherwise drop the session the codec
382-
/// still points at. Retaining ancestors here rather than on the codec
383-
/// avoids a cycle: nothing reachable from a `SessionContext` reaches back
384-
/// to a `PySessionContext`.
385-
ancestors: Vec<Arc<SessionContext>>,
386376
}
387377

388378
#[pymethods]
@@ -415,7 +405,6 @@ impl PySessionContext {
415405
ctx,
416406
logical_codec: Arc::new(PythonLogicalCodec::default()),
417407
physical_codec: Arc::new(PythonPhysicalCodec::default()),
418-
ancestors: Vec::new(),
419408
})
420409
}
421410

@@ -424,7 +413,6 @@ impl PySessionContext {
424413
ctx: Arc::new(self.ctx.as_ref().clone().enable_url_table()),
425414
logical_codec: Arc::clone(&self.logical_codec),
426415
physical_codec: Arc::clone(&self.physical_codec),
427-
ancestors: self.ancestors_including_self(),
428416
})
429417
}
430418

@@ -436,7 +424,6 @@ impl PySessionContext {
436424
ctx,
437425
logical_codec: Arc::new(PythonLogicalCodec::default()),
438426
physical_codec: Arc::new(PythonPhysicalCodec::default()),
439-
ancestors: Vec::new(),
440427
})
441428
}
442429

@@ -1231,14 +1218,16 @@ impl PySessionContext {
12311218
) -> PyDataFusionResult<Self> {
12321219
let planner = ffi_query_planner_from_pycapsule(&planner, Some(slf.as_any()))?;
12331220
let this = slf.borrow();
1234-
let ctx =
1235-
this.ctx_with_rebound_planner(&this.logical_codec, &this.physical_codec, Some(planner));
1221+
let (ctx, logical_codec, physical_codec) = this.derived_parts(
1222+
Arc::clone(&this.logical_codec),
1223+
Arc::clone(&this.physical_codec),
1224+
Some(planner),
1225+
);
12361226

12371227
Ok(Self {
1238-
ancestors: this.ancestors_for(&ctx),
12391228
ctx,
1240-
logical_codec: Arc::clone(&this.logical_codec),
1241-
physical_codec: Arc::clone(&this.physical_codec),
1229+
logical_codec,
1230+
physical_codec,
12421231
})
12431232
}
12441233

@@ -1452,10 +1441,9 @@ impl PySessionContext {
14521441
let logical_codec = Arc::new(PythonLogicalCodec::new(inner));
14531442

14541443
let this = slf.borrow();
1455-
let physical_codec = Arc::clone(&this.physical_codec);
1456-
let ctx = this.ctx_with_rebound_planner(&logical_codec, &physical_codec, None);
1444+
let (ctx, logical_codec, physical_codec) =
1445+
this.derived_parts(logical_codec, Arc::clone(&this.physical_codec), None);
14571446
Ok(Self {
1458-
ancestors: this.ancestors_for(&ctx),
14591447
ctx,
14601448
logical_codec,
14611449
physical_codec,
@@ -1482,10 +1470,9 @@ impl PySessionContext {
14821470
let physical_codec = Arc::new(PythonPhysicalCodec::new(inner));
14831471

14841472
let this = slf.borrow();
1485-
let logical_codec = Arc::clone(&this.logical_codec);
1486-
let ctx = this.ctx_with_rebound_planner(&logical_codec, &physical_codec, None);
1473+
let (ctx, logical_codec, physical_codec) =
1474+
this.derived_parts(Arc::clone(&this.logical_codec), physical_codec, None);
14871475
Ok(Self {
1488-
ancestors: this.ancestors_for(&ctx),
14891476
ctx,
14901477
logical_codec,
14911478
physical_codec,
@@ -1501,9 +1488,9 @@ impl PySessionContext {
15011488
PythonPhysicalCodec::new(Arc::clone(self.physical_codec.inner()))
15021489
.with_python_udf_inlining(enabled),
15031490
);
1504-
let ctx = self.ctx_with_rebound_planner(&logical_codec, &physical_codec, None);
1491+
let (ctx, logical_codec, physical_codec) =
1492+
self.derived_parts(logical_codec, physical_codec, None);
15051493
Self {
1506-
ancestors: self.ancestors_for(&ctx),
15071494
ctx,
15081495
logical_codec,
15091496
physical_codec,
@@ -1512,50 +1499,35 @@ impl PySessionContext {
15121499
}
15131500

15141501
impl PySessionContext {
1515-
/// Keep-alive list for a context derived from this one.
1516-
///
1517-
/// Sharing the same `SessionContext` carries the list through unchanged.
1518-
/// A fork adds this session, because a foreign codec installed here holds
1519-
/// it weakly and would otherwise dangle once the Python name is rebound.
1520-
fn ancestors_for(&self, derived: &Arc<SessionContext>) -> Vec<Arc<SessionContext>> {
1521-
if Arc::ptr_eq(derived, &self.ctx) {
1522-
self.ancestors.clone()
1523-
} else {
1524-
self.ancestors_including_self()
1525-
}
1526-
}
1527-
1528-
fn ancestors_including_self(&self) -> Vec<Arc<SessionContext>> {
1529-
let mut ancestors = self.ancestors.clone();
1530-
ancestors.push(Arc::clone(&self.ctx));
1531-
ancestors
1532-
}
1533-
1534-
/// Return the [`SessionContext`] a derived `PySessionContext` should use,
1535-
/// binding a foreign query planner to `logical_codec` and `physical_codec`.
1502+
/// Return the pieces a derived `PySessionContext` should hold, binding any
1503+
/// foreign query planner and foreign codecs to the session that will run
1504+
/// the query.
15361505
///
15371506
/// Pass `Some(planner)` to install one, or `None` to rebind whichever
15381507
/// planner the session already holds.
15391508
///
15401509
/// With no foreign planner in play there is nothing to rebind, so the
1541-
/// existing context is shared and swapping codecs alone does not change
1542-
/// what a derived context observes. A foreign planner does have to be
1543-
/// rebound, and the FFI codecs capture the context they are built against,
1544-
/// so that case forks the session: the codecs are built from the fork and
1545-
/// then the fork's state is overwritten in place, because rebuilding the
1546-
/// context afterwards would leave the codecs pointing at a session that is
1547-
/// no longer used for planning.
1510+
/// existing context and codecs are shared unchanged. A foreign planner does
1511+
/// have to be rebound, and that forks the session, because installing it
1512+
/// writes to `SessionState` and the receiver must not be modified. The
1513+
/// codecs are built from the fork and the fork's state is then overwritten
1514+
/// in place, since rebuilding the context afterwards would leave them
1515+
/// pointing at a session that is no longer used for planning.
15481516
///
15491517
/// A fork is not a deep copy. `SessionState` keeps its catalog list behind
15501518
/// an `Arc`, so catalogs and tables stay shared with the original session,
15511519
/// while registered functions and the configuration are snapshotted at the
15521520
/// time of the call. The session id is deliberately carried over.
1553-
fn ctx_with_rebound_planner(
1521+
fn derived_parts(
15541522
&self,
1555-
logical_codec: &Arc<PythonLogicalCodec>,
1556-
physical_codec: &Arc<PythonPhysicalCodec>,
1523+
logical_codec: Arc<PythonLogicalCodec>,
1524+
physical_codec: Arc<PythonPhysicalCodec>,
15571525
planner: Option<FFI_QueryPlanner>,
1558-
) -> Arc<SessionContext> {
1526+
) -> (
1527+
Arc<SessionContext>,
1528+
Arc<PythonLogicalCodec>,
1529+
Arc<PythonPhysicalCodec>,
1530+
) {
15591531
let state = self.ctx.state();
15601532

15611533
// When the caller is only replacing codecs, recover the handle behind
@@ -1570,23 +1542,66 @@ impl PySessionContext {
15701542
});
15711543

15721544
let Some(planner) = planner else {
1573-
return Arc::clone(&self.ctx);
1545+
return (Arc::clone(&self.ctx), logical_codec, physical_codec);
15741546
};
15751547

15761548
let ctx = Arc::new(SessionContext::new_with_state(state));
1549+
let logical_codec = Arc::new(Self::rebound_logical_codec(&ctx, &logical_codec));
1550+
let physical_codec = Arc::new(Self::rebound_physical_codec(&ctx, &physical_codec));
1551+
15771552
let inner: Arc<dyn QueryPlanner + Send + Sync> = (&planner).into();
15781553
let planner: Arc<dyn QueryPlanner + Send + Sync> = (&FFI_QueryPlanner::new_with_ffi_codecs(
15791554
inner,
1580-
Self::ffi_logical_codec_for(&ctx, logical_codec),
1581-
Self::ffi_physical_codec_for(&ctx, physical_codec),
1555+
Self::ffi_logical_codec_for(&ctx, &logical_codec),
1556+
Self::ffi_physical_codec_for(&ctx, &physical_codec),
15821557
))
15831558
.into();
15841559
let state = SessionStateBuilder::new_from_existing(ctx.state())
15851560
.with_query_planner(planner)
15861561
.build();
15871562
*ctx.state_ref().write() = state;
15881563

1589-
ctx
1564+
(ctx, logical_codec, physical_codec)
1565+
}
1566+
1567+
/// Rebind a foreign logical codec to `ctx`.
1568+
///
1569+
/// A codec imported from another library holds an `FFI_TaskContextProvider`
1570+
/// pointing at whichever session it was installed on. Its decode callbacks
1571+
/// resolve names against that session, so a fork has to move them onto the
1572+
/// fork or they keep answering from the pre-fork registry.
1573+
///
1574+
/// `FFI_LogicalExtensionCodec::new` adopts the provider supplied here when
1575+
/// the codec is already foreign, returning a *clone* of the handle, so the
1576+
/// context this codec was derived from keeps its own binding. A codec this
1577+
/// library owns has no stored provider, and round-trips back to the same
1578+
/// `Arc` unchanged, so this is safe to apply unconditionally.
1579+
fn rebound_logical_codec(
1580+
ctx: &Arc<SessionContext>,
1581+
codec: &Arc<PythonLogicalCodec>,
1582+
) -> PythonLogicalCodec {
1583+
let ctx_provider = Arc::clone(ctx) as Arc<dyn TaskContextProvider>;
1584+
let runtime = get_tokio_runtime().handle().clone();
1585+
let rebound =
1586+
FFI_LogicalExtensionCodec::new(Arc::clone(codec.inner()), Some(runtime), &ctx_provider);
1587+
PythonLogicalCodec::new((&rebound).into())
1588+
.with_python_udf_inlining(codec.python_udf_inlining())
1589+
}
1590+
1591+
/// Physical counterpart of [`Self::rebound_logical_codec`].
1592+
fn rebound_physical_codec(
1593+
ctx: &Arc<SessionContext>,
1594+
codec: &Arc<PythonPhysicalCodec>,
1595+
) -> PythonPhysicalCodec {
1596+
let ctx_provider = Arc::clone(ctx) as Arc<dyn TaskContextProvider>;
1597+
let runtime = get_tokio_runtime().handle().clone();
1598+
let rebound = FFI_PhysicalExtensionCodec::new(
1599+
Arc::clone(codec.inner()),
1600+
Some(runtime),
1601+
&ctx_provider,
1602+
);
1603+
PythonPhysicalCodec::new((&rebound).into())
1604+
.with_python_udf_inlining(codec.python_udf_inlining())
15901605
}
15911606

15921607
async fn _table(&self, name: &str) -> datafusion::common::Result<DataFrame> {
@@ -1816,7 +1831,6 @@ impl From<SessionContext> for PySessionContext {
18161831
ctx: Arc::new(ctx),
18171832
logical_codec: Arc::new(PythonLogicalCodec::default()),
18181833
physical_codec: Arc::new(PythonPhysicalCodec::default()),
1819-
ancestors: Vec::new(),
18201834
}
18211835
}
18221836
}

‎docs/source/contributor-guide/ffi.md‎

Lines changed: 19 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -294,19 +294,27 @@ satisfies the same protocol an extension library implements. When you export the
294294
planner to wrap it, `ctx.__datafusion_query_planner__()` and
295295
`ctx.__datafusion_query_planner__(ctx)` are both fine.
296296

297-
### A codec decodes against the session it was installed on
297+
### A codec decodes against the session that is running the query
298298

299299
Because the provider comes from the host, a decode callback running inside an extension
300-
library resolves names against the session that is running the query. A function
301-
registered with `ctx.register_udf(...)` is visible to a foreign codec decoding a node
302-
that references it by name, and the handle is live rather than a snapshot, so a
303-
registration made after the codec is installed is visible too.
304-
305-
The one boundary is the fork. Installing a foreign query planner forks the session, and
306-
a codec installed beforehand keeps pointing at the session it was built against, so a
307-
function registered after that point is not visible to it. This is the same rule as the
308-
caveat below, seen from the codec's side: register before deriving. Both halves are
309-
covered in
300+
library resolves names against the session running the query. A function registered with
301+
`ctx.register_udf(...)` is visible to a foreign codec decoding a node that references it
302+
by name, and the handle is live rather than a snapshot, so a registration made after the
303+
codec is installed is visible too.
304+
305+
This survives a fork. Installing a foreign query planner forks the session, and the fork
306+
rebinds every foreign codec it carries onto the new session, so a function registered
307+
after the fork is still visible to them. `FFI_LogicalExtensionCodec::new` adopts the
308+
provider supplied to it when the codec is already foreign, returning a clone of the
309+
handle, so the context the fork was derived from keeps its own binding and continues to
310+
resolve against its own session.
311+
312+
That behaviour requires DataFusion 55.1.0 or newer. Before it, those constructors
313+
silently discarded the provider, a foreign codec could not be rebound, and a fork left
314+
it resolving against the pre-fork session
315+
([apache/datafusion#24722](https://github.com/apache/datafusion/issues/24722)).
316+
317+
All three properties are covered in
310318
`examples/datafusion-ffi-query-planner-example/python/tests/_test_three_library_query_planner.py`,
311319
where the example codecs take a `require_udf_on_decode` name and resolve it out of the
312320
task context they are handed.

0 commit comments

Comments
 (0)