Skip to content

Commit 7789bb2

Browse files
timsaucerclaude
andcommitted
Name the execute partition index for what it is, trim the upgrade guide
`SessionContext.execute` took its second argument as `partitions`, which reads as a count when it is a single partition index. Rename it to `partition` and give the method a real docstring with a doctest covering both a full sweep over `partition_count` and the out-of-range `ValueError`. Every call site in the repo, docs, and examples passes it positionally, so add a short note to the upgrade guide for anyone passing it by keyword. Drop two changelog-shaped sections from the upgrade guide. `output_partitioning` is additive and `SessionContext.execute` / `SessionConfig.set` only trade a panic for a raise, so neither asks the reader to change anything. The `with_extensions` recommendation goes for the same reason; it is advice, and the extension guide already carries it under `extension_bundles`. Also remove the comment above the partition range check. It explained the check by way of a `MemorySourceConfig` panic, which reads as a complaint about DataFusion's leaves. The index arrives straight from Python and no planner has seen it, so validating it needs no more justification than any other argument check at the boundary. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1 parent a26b90a commit 7789bb2

4 files changed

Lines changed: 43 additions & 53 deletions

File tree

‎crates/core/src/context.rs‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1422,10 +1422,6 @@ impl PySessionContext {
14221422
) -> PyDataFusionResult<PyRecordBatchStream> {
14231423
let ctx: TaskContext = TaskContext::from(&self.ctx.state());
14241424
let plan = plan.plan.clone();
1425-
// Checked here because the leaves index their partitions directly: a
1426-
// `MemorySourceConfig` panics with a bare `index out of bounds`, which
1427-
// surfaces as a `JoinError::Panic` naming neither the plan nor the
1428-
// partition the caller asked for.
14291425
let partition_count = plan.output_partitioning().partition_count();
14301426
if part >= partition_count {
14311427
return Err(PyValueError::new_err(format!(

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

Lines changed: 6 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -96,16 +96,6 @@ way `add_physical_optimizer_rule` does and returns nothing — the query planner
9696
lives in `SessionState`, so it belongs to the session rather than to a
9797
particular handle on it. See {ref}`extension_planners` for the full protocol.
9898

99-
If a library ships codecs *and* a planner, prefer
100-
`SessionContext.with_extensions(bundle)` over installing each piece by hand. It
101-
installs every codec before it binds any planner, so a planner cannot end up
102-
carrying a chain that a later `with_logical_extension_codec` call has grown.
103-
The library exposes a bundle object implementing
104-
`__datafusion_session_components__` for its codecs and
105-
`__datafusion_session_planner__` for its planner — the latter is handed the
106-
planner installed so far, so several libraries that each ship one nest instead
107-
of displacing each other. See {ref}`extension_bundles`.
108-
10999
(extension_version_mismatch)=
110100

111101
### Mismatched extension libraries now fail loudly
@@ -169,41 +159,17 @@ installed produces the same bytes as before, as do functions encoded by name.
169159
Regenerate any plan you serialized with an earlier release and stored for later
170160
use, if it was produced by a session with an extension codec installed.
171161

172-
### Physical plans report their partitioning scheme
162+
### `SessionContext.execute` renamed its second parameter
173163

174-
{py:attr}`~datafusion.ExecutionPlan.output_partitioning` is new, and reports
175-
what {py:attr}`~datafusion.ExecutionPlan.partition_count` leaves out: whether
176-
the rows in those partitions are hash-distributed on known keys, spread
177-
round-robin, or merely counted. It returns a
178-
{py:class}`~datafusion.PhysicalPartitioning`.
164+
The parameter is a single partition index, not a count, and is now named
165+
`partition` rather than `partitions`. Positional calls are unaffected; update
166+
any call passing it by keyword.
179167

180168
```python
181-
partitioning = df.execution_plan().output_partitioning
182-
partitioning.scheme # 'Hash', 'RoundRobinBatch', 'Range', 'UnknownPartitioning'
183-
partitioning.partition_count
184-
partitioning.hash_expressions # display strings, or None
169+
ctx.execute(plan, partitions=0) # before
170+
ctx.execute(plan, partition=0) # after
185171
```
186172

187-
This is additive; `partition_count` keeps working and agrees with
188-
`output_partitioning.partition_count`. Note that
189-
{py:class}`datafusion.expr.Partitioning` is a different type: that one is the
190-
*logical* partitioning {py:meth}`~datafusion.DataFrame.repartition_by_hash`
191-
takes as a request, while this one is what a built plan actually does.
192-
193-
### Two error paths that used to abort the interpreter
194-
195-
{py:meth}`~datafusion.SessionContext.execute` now raises `ValueError` for a
196-
partition index that is out of range. Previously the plan's leaves indexed
197-
their partitions directly and the resulting Rust panic surfaced as an error
198-
naming neither the plan nor the index.
199-
200-
{py:meth}`~datafusion.SessionConfig.set` now raises for a key whose namespace
201-
does not exist -- `datafusion.runtime.*`, or a config extension that has not
202-
been installed yet. Previously it aborted with a `PanicException`, which
203-
derives from `BaseException` and so escaped `except Exception`. This matters
204-
when replaying settings read back from `information_schema.df_settings`, which
205-
lists keys in both of those categories.
206-
207173
### Changes to the `datafusion-python-util` crate
208174

209175
Extension libraries written in Rust usually depend on the

‎python/datafusion/context.py‎

Lines changed: 36 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2351,9 +2351,42 @@ def read_table(
23512351
"""Creates a :py:class:`~datafusion.dataframe.DataFrame` from a table."""
23522352
return DataFrame(self.ctx.read_table(table))
23532353

2354-
def execute(self, plan: ExecutionPlan, partitions: int) -> RecordBatchStream:
2355-
"""Execute the ``plan`` and return the results."""
2356-
return RecordBatchStream(self.ctx.execute(plan._raw_plan, partitions))
2354+
def execute(self, plan: ExecutionPlan, partition: int) -> RecordBatchStream:
2355+
"""Execute a single partition of ``plan`` and stream its batches.
2356+
2357+
Args:
2358+
plan: The physical plan to execute.
2359+
partition: Index of the partition to execute, in
2360+
``range(plan.partition_count)``.
2361+
2362+
Returns:
2363+
A stream over the record batches that partition produces.
2364+
2365+
Raises:
2366+
ValueError: If ``partition`` is not a valid index for ``plan``.
2367+
2368+
Example usage:
2369+
2370+
>>> import pyarrow as pa
2371+
>>> from datafusion import SessionContext
2372+
>>> ctx = SessionContext()
2373+
>>> ctx.register_record_batches(
2374+
... "t", [[pa.record_batch({"a": [1, 2]})], [pa.record_batch({"a": [3]})]]
2375+
... )
2376+
>>> plan = ctx.sql("select a from t").execution_plan()
2377+
>>> plan.partition_count
2378+
2
2379+
>>> sum(
2380+
... batch.to_pyarrow().num_rows
2381+
... for p in range(plan.partition_count)
2382+
... for batch in ctx.execute(plan, p)
2383+
... )
2384+
3
2385+
>>> ctx.execute(plan, 2)
2386+
Traceback (most recent call last):
2387+
ValueError: Partition index 2 is out of range for a plan with 2 partition(s)
2388+
"""
2389+
return RecordBatchStream(self.ctx.execute(plan._raw_plan, partition))
23572390

23582391
@staticmethod
23592392
def _convert_file_sort_order(

‎python/tests/test_plans.py‎

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -145,12 +145,7 @@ def test_output_partitioning_reports_the_scheme_not_just_the_count() -> None:
145145

146146

147147
def test_execute_rejects_an_out_of_range_partition() -> None:
148-
"""An out-of-range partition index raises instead of panicking.
149-
150-
The leaves index their partition vector directly, so without this check a
151-
bad index surfaces as a `JoinError::Panic` carrying `index out of bounds`
152-
and naming neither the plan nor the index requested.
153-
"""
148+
"""An out-of-range partition index raises instead of panicking."""
154149
ctx = SessionContext()
155150
ctx.register_record_batches("t", [[pa.record_batch({"a": [1, 2, 3]})]])
156151
plan = ctx.sql("select a from t").execution_plan()

0 commit comments

Comments
 (0)