Skip to content

Commit 249d5fb

Browse files
committed
Fix enable_url_table session sharing
1 parent 878e5d3 commit 249d5fb

10 files changed

Lines changed: 99 additions & 41 deletions

File tree

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

Lines changed: 5 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -187,13 +187,11 @@ an instruction to derive one first. It is not: the factories are handed the
187187
receiver, and the returned handle shares its allocation. There is nothing to
188188
keep alive separately and nothing to garbage-collect out from under a provider.
189189

190-
`SessionContext.enable_url_table` is the one method that mints a second
191-
allocation for a session. Its result must not outlive the receiver, and it also
192-
forks the session's `SessionState` while keeping its id, so two handles report
193-
one `session_id()` with divergent configuration. That is a bug rather than a
194-
design — tracked in
195-
[apache/datafusion-python#1708](https://github.com/apache/datafusion-python/issues/1708)
196-
— so do not cite it as precedent for deriving a replacement context.
190+
`SessionContext.enable_url_table` follows this rule: it rebuilds the catalog
191+
inside the receiver's existing `SessionState` and returns another handle
192+
sharing the same `Arc<SessionContext>`. It preserves the session id and keeps
193+
weak FFI providers valid. An idempotence guard avoids wrapping the dynamic-file
194+
catalog more than once.
197195

198196
## Rule 7 — installing a planner mutates the session, and says so
199197

‎Cargo.lock‎

Lines changed: 1 addition & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎crates/core/Cargo.toml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@ datafusion = { workspace = true, features = ["avro", "unicode_expressions"] }
5353
datafusion-substrait = { workspace = true, optional = true }
5454
datafusion-proto = { workspace = true }
5555
datafusion-ffi = { workspace = true }
56+
datafusion-session = { workspace = true }
5657
datafusion-spark = { workspace = true, features = ["core"] }
5758
prost = { workspace = true } # keep in line with `datafusion-substrait`
5859
serde_json = { workspace = true }

‎crates/core/src/context.rs‎

Lines changed: 34 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -27,8 +27,11 @@ use arrow::pyarrow::FromPyArrow;
2727
use datafusion::arrow::datatypes::{DataType, Schema, SchemaRef};
2828
use datafusion::arrow::pyarrow::PyArrowType;
2929
use datafusion::arrow::record_batch::RecordBatch;
30-
use datafusion::catalog::{CatalogProvider, CatalogProviderList, TableProviderFactory};
30+
use datafusion::catalog::{
31+
CatalogProvider, CatalogProviderList, DynamicFileCatalog, TableProviderFactory, UrlTableFactory,
32+
};
3133
use datafusion::common::{DFSchema, ScalarValue, TableReference, exec_err};
34+
use datafusion::datasource::dynamic_file::DynamicListTableFactory;
3235
use datafusion::datasource::file_format::file_compression_type::FileCompressionType;
3336
use datafusion::datasource::file_format::parquet::ParquetFormat;
3437
use datafusion::datasource::listing::{
@@ -65,6 +68,7 @@ use datafusion_python_util::{
6568
ffi_query_planner_from_pycapsule, get_global_ctx, get_tokio_runtime,
6669
physical_optimizer_rule_from_pycapsule, spawn_future, validate_pycapsule, wait_for_future,
6770
};
71+
use datafusion_session::SessionStore;
6872
use object_store::ObjectStore;
6973
use pyo3::IntoPyObjectExt;
7074
use pyo3::exceptions::{PyKeyError, PyRuntimeError, PyTypeError, PyValueError};
@@ -452,15 +456,30 @@ impl PySessionContext {
452456
}
453457

454458
pub fn enable_url_table(&self) -> PyResult<Self> {
455-
// Pre-existing caveat, unrelated to query planners: this is the one
456-
// method that mints a second `Arc<SessionContext>` for a session, and
457-
// it also forks the session's state while keeping its id. Any weak
458-
// `FFI_TaskContextProvider` handed out by the receiver stays bound to
459-
// the receiver, so the returned context must not outlive it. See
460-
// `set_session_query_planner` for why everything else mutates in place.
461-
// Tracked as a bug in <https://github.com/apache/datafusion-python/issues/1708>.
459+
let state_ref = self.ctx.state_ref();
460+
let mut guard = state_ref.write();
461+
if guard
462+
.catalog_list()
463+
.downcast_ref::<DynamicFileCatalog>()
464+
.is_none()
465+
{
466+
let factory = Arc::new(DynamicListTableFactory::new(SessionStore::new()));
467+
let catalog_list = Arc::new(DynamicFileCatalog::new(
468+
Arc::clone(guard.catalog_list()),
469+
Arc::clone(&factory) as Arc<dyn UrlTableFactory>,
470+
));
471+
factory
472+
.session_store()
473+
.with_state(self.ctx.state_weak_ref());
474+
475+
*guard = SessionStateBuilder::new_from_existing(guard.clone())
476+
.with_session_id(guard.session_id().to_string())
477+
.with_catalog_list(catalog_list)
478+
.build();
479+
}
480+
462481
Ok(PySessionContext {
463-
ctx: Arc::new(self.ctx.as_ref().clone().enable_url_table()),
482+
ctx: Arc::clone(&self.ctx),
464483
logical_codec: Arc::clone(&self.logical_codec),
465484
physical_codec: Arc::clone(&self.physical_codec),
466485
})
@@ -1471,14 +1490,12 @@ impl PySessionContext {
14711490
/// time, and a payload written through one would resolve to the other on
14721491
/// decode. See [`SESSION_CODEC_ID_PREFIX`].
14731492
///
1474-
/// Handles derived from one session — `with_python_udf_inlining`,
1475-
/// `with_logical_extension_codec`, [`Self::_install_extension_codecs`] —
1476-
/// report the same id even though their codec chains differ, so installing
1477-
/// two of them on one target is refused. That is the intended answer: they
1478-
/// share a `state_ref`, so their payloads would resolve against the same
1479-
/// session and are indistinguishable on decode. Every derivation shares the
1480-
/// session for exactly this reason; `enable_url_table` is the one that does
1481-
/// not, and it is tracked as a bug.
1493+
/// Handles derived from one session — including `with_python_udf_inlining`,
1494+
/// `with_logical_extension_codec`, [`Self::_install_extension_codecs`], and
1495+
/// `enable_url_table` — report the same id even though their codec chains
1496+
/// differ, so installing two of them on one target is refused. That is the
1497+
/// intended answer: they share a `state_ref`, so their payloads would
1498+
/// resolve against the same session and are indistinguishable on decode.
14821499
#[getter]
14831500
pub fn __datafusion_codec_id__(&self) -> String {
14841501
format!("{SESSION_CODEC_ID_PREFIX}{}", self.ctx.session_id())

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

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -64,13 +64,12 @@ would close the cycle
6464
`SessionContext -> catalog -> FFI provider -> FFI codec -> SessionContext` and
6565
leak it.
6666

67-
`SessionContext.enable_url_table` is the one exception. It clones the
68-
underlying `SessionContext`, so the returned context has an allocation of its
69-
own and must not outlive the receiver. It also forks the session's state while
70-
keeping its id, so two handles report one `session_id()` with divergent
71-
configuration. That is a bug rather than a design, tracked in
72-
[apache/datafusion-python#1708](https://github.com/apache/datafusion-python/issues/1708);
73-
do not copy the pattern.
67+
`SessionContext.enable_url_table` follows the same rule. It installs the URL
68+
catalog wrapper by rebuilding `SessionState` through the existing context's
69+
`state_ref()`, then returns another handle sharing the same
70+
`Arc<SessionContext>`. The session id and weak FFI task-context providers remain
71+
valid across the derivation. Repeated calls detect the existing dynamic-file
72+
catalog and do not nest another wrapper.
7473

7574
(ffi_internals_rebinding)=
7675

‎docs/source/extension-guide/sessions.md‎

Lines changed: 4 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -105,13 +105,8 @@ The same rule applies to a capsule you take off a context inside your own code:
105105
a codec capsule taken from a throwaway `SessionContext()` names a session that
106106
is already gone and fails on first use.
107107

108-
:::{warning}
109-
{py:meth}`~datafusion.SessionContext.enable_url_table` is an exception to the
110-
one-session-one-allocation rule above: it clones the underlying
111-
`SessionContext`, so the returned context has an allocation of its own and must
112-
not outlive the receiver. It also forks the session's state while keeping its
113-
id, so two handles report one `session_id()` with divergent configuration. That
114-
is a bug rather than a design, tracked in
115-
[apache/datafusion-python#1708](https://github.com/apache/datafusion-python/issues/1708);
116-
do not build on the behaviour.
108+
:::{note}
109+
{py:meth}`~datafusion.SessionContext.enable_url_table` returns another handle
110+
to the same session and enables URL tables for every handle on it. Repeated
111+
calls are safe and do not wrap the catalog more than once.
117112
:::

‎docs/source/user-guide/upgrade-guides.md‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,22 @@
2121

2222
## DataFusion 56.0.0
2323

24+
### `SessionContext.enable_url_table` now shares the existing session
25+
26+
`enable_url_table()` now returns a handle on the same session rather than a
27+
separate session with copied state. The original handle sees URL tables enabled
28+
too, and state changes remain visible through both handles. Code that reassigns
29+
the result continues to work unchanged:
30+
31+
```python
32+
ctx = SessionContext()
33+
ctx = ctx.enable_url_table()
34+
```
35+
36+
Before this change, the two handles could report the same session id while
37+
holding divergent state; replacing the original handle could also invalidate
38+
FFI providers bound to it. Treat both handles as views of one session.
39+
2440
### `sort` and `order_by` now order `NULLS LAST` by default
2541

2642
Calling `.sort(...)` on a DataFrame or using the `.order_by(...)` function now

‎examples/datafusion-ffi-example/python/tests/_test_table_provider.py‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,3 +51,19 @@ def test_table_provider_ffi(inner_capsule: bool) -> None:
5151
result = ctx.read_table(table).collect()
5252
result = sorted((r.column(0) for r in result), key=lambda a: a[0].as_py())
5353
assert result == expected
54+
55+
56+
def test_table_provider_survives_enable_url_table():
57+
ctx = SessionContext()
58+
ctx.register_table("t", MyTableProvider(3, 2, 4))
59+
session_id = ctx.session_id()
60+
61+
# Reassigning the derived handle must not orphan FFI providers tied to the
62+
# session, and enabling URL tables repeatedly must remain safe.
63+
ctx = ctx.enable_url_table()
64+
ctx = ctx.enable_url_table()
65+
66+
assert ctx.session_id() == session_id
67+
result = ctx.table("t").collect()
68+
assert len(result) == 4
69+
assert result[0].num_columns == 3

‎python/datafusion/context.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -669,7 +669,7 @@ def enable_url_table(self) -> SessionContext:
669669
"""Control if local files can be queried as tables.
670670
671671
Returns:
672-
A new :py:class:`SessionContext` object with url table enabled.
672+
A handle on the same session with url table enabled.
673673
"""
674674
klass = self.__class__
675675
obj = klass.__new__(klass)

‎python/tests/test_context.py‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -939,6 +939,21 @@ def test_contexts_sharing_a_session_share_the_planner(ctx):
939939
assert sibling.session_id() == ctx.session_id()
940940

941941

942+
def test_enable_url_table_returns_handle_on_shared_session(ctx):
943+
sibling = ctx.enable_url_table()
944+
session_id = ctx.session_id()
945+
946+
ctx.sql("SET datafusion.execution.batch_size = 111").collect()
947+
948+
assert sibling.session_id() == session_id
949+
assert ctx.sql("SHOW datafusion.execution.batch_size").collect()[0].column(
950+
1
951+
).to_pylist() == ["111"]
952+
assert sibling.sql("SHOW datafusion.execution.batch_size").collect()[0].column(
953+
1
954+
).to_pylist() == ["111"]
955+
956+
942957
class _NamedCodec:
943958
"""Wraps a codec capsule in an object that can name itself.
944959

0 commit comments

Comments
 (0)