@@ -44,6 +44,7 @@ use datafusion::execution::options::{ArrowReadOptions, ReadOptions};
 use datafusion::execution::runtime_env::RuntimeEnvBuilder;
 use datafusion::execution::session_state::SessionStateBuilder;
 use datafusion::execution::{FunctionRegistry, TaskContextProvider};
+use datafusion::physical_plan::ExecutionPlanProperties;
 use datafusion::prelude::{
     AvroReadOptions, CsvReadOptions, DataFrame, JsonReadOptions, ParquetReadOptions,
 };
@@ -193,8 +194,17 @@ impl PySessionConfig {
         Self::from(self.config.clone().with_parquet_pruning(enabled))
     }
 
-    fn set(&self, key: &str, value: &str) -> Self {
-        Self::from(self.config.clone().set_str(key, value))
+    /// Set a config option by key.
+    ///
+    /// Not routed through `SessionConfig::set_str`, which unwraps the result:
+    /// an unknown namespace -- `datafusion.runtime.*`, or a config extension
+    /// that has not been installed yet -- would abort as a `PanicException`
+    /// rather than raise. `information_schema.df_settings` lists keys in both
+    /// of those categories, so replaying it is otherwise unsafe.
+    fn set(&self, key: &str, value: &str) -> PyDataFusionResult<Self> {
+        let mut config = self.config.clone();
+        config.options_mut().set(key, value)?;
+        Ok(Self::from(config))
     }
 
     pub fn with_extension(&self, extension: Bound<PyAny>) -> PyResult<Self> {
@@ -1412,6 +1422,18 @@ impl PySessionContext {
     ) -> PyDataFusionResult<PyRecordBatchStream> {
         let ctx: TaskContext = TaskContext::from(&self.ctx.state());
         let plan = plan.plan.clone();
+        // Checked here because the leaves index their partitions directly: a
+        // `MemorySourceConfig` panics with a bare `index out of bounds`, which
+        // surfaces as a `JoinError::Panic` naming neither the plan nor the
+        // partition the caller asked for.
+        let partition_count = plan.output_partitioning().partition_count();
+        if part >= partition_count {
+            return Err(PyValueError::new_err(format!(
+                "Partition index {part} is out of range for a plan with \
+                 {partition_count} partition(s)"
+            ))
+            .into());
+        }
         let stream = spawn_future(py, async move { plan.execute(part, Arc::new(ctx)) })?;
         Ok(PyRecordBatchStream::new(stream))
     }
 
@@ -94,6 +94,7 @@ fn _internal(py: Python, m: Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<metrics::PyMetricsSet>()?;
     m.add_class::<metrics::PyMetric>()?;
     m.add_class::<physical_plan::PyExecutionPlan>()?;
+    m.add_class::<physical_plan::PyPhysicalPartitioning>()?;
     m.add_class::<record_batch::PyRecordBatch>()?;
     m.add_class::<record_batch::PyRecordBatchStream>()?;
 
 
@@ -17,6 +17,7 @@
 
 use std::sync::Arc;
 
+use datafusion::physical_expr::Partitioning;
 use datafusion::physical_plan::{ExecutionPlan, ExecutionPlanProperties, displayable};
 use datafusion_proto::physical_plan::AsExecutionPlan;
 use prost::Message;
@@ -126,6 +127,82 @@ impl PyExecutionPlan {
     pub fn partition_count(&self) -> usize {
         self.plan.output_partitioning().partition_count()
     }
+
+    #[getter]
+    pub fn output_partitioning(&self) -> PyPhysicalPartitioning {
+        self.plan.output_partitioning().clone().into()
+    }
+}
+
+/// How a physical plan's output rows are spread across its partitions.
+///
+/// Distinct from `datafusion.expr.Partitioning`, which is the *logical*
+/// partitioning `DataFrame.repartition` takes as a request. This one reports
+/// what a built plan actually does.
+#[pyclass(
+    from_py_object,
+    frozen,
+    name = "PhysicalPartitioning",
+    module = "datafusion",
+    subclass
+)]
+#[derive(Debug, Clone)]
+pub struct PyPhysicalPartitioning {
+    partitioning: Partitioning,
+}
+
+#[pymethods]
+impl PyPhysicalPartitioning {
+    /// Which partitioning scheme this is.
+    ///
+    /// One of `RoundRobinBatch`, `Hash`, `Range`, or `UnknownPartitioning`.
+    /// `UnknownPartitioning` is what a plan reports when it knows how many
+    /// partitions it has but nothing about how rows are distributed between
+    /// them, which is the common case for a file scan.
+    #[getter]
+    pub fn scheme(&self) -> &'static str {
+        match self.partitioning {
+            Partitioning::RoundRobinBatch(_) => "RoundRobinBatch",
+            Partitioning::Hash(_, _) => "Hash",
+            Partitioning::Range(_) => "Range",
+            Partitioning::UnknownPartitioning(_) => "UnknownPartitioning",
+        }
+    }
+
+    #[getter]
+    pub fn partition_count(&self) -> usize {
+        self.partitioning.partition_count()
+    }
+
+    /// The expressions rows are hashed on, or `None` for other schemes.
+    ///
+    /// These are physical expressions, which have no Python representation, so
+    /// they are returned in their displayed form.
+    #[getter]
+    pub fn hash_expressions(&self) -> Option<Vec<String>> {
+        match &self.partitioning {
+            Partitioning::Hash(exprs, _) => {
+                Some(exprs.iter().map(|expr| format!("{expr}")).collect())
+            }
+            _ => None,
+        }
+    }
+
+    fn __repr__(&self) -> String {
+        format!("{}", self.partitioning)
+    }
+}
+
+impl From<Partitioning> for PyPhysicalPartitioning {
+    fn from(partitioning: Partitioning) -> Self {
+        Self { partitioning }
+    }
+}
+
+impl From<PyPhysicalPartitioning> for Partitioning {
+    fn from(partitioning: PyPhysicalPartitioning) -> Self {
+        partitioning.partitioning
+    }
 }
 
 impl From<PyExecutionPlan> for Arc<dyn ExecutionPlan> {
 
@@ -169,6 +169,41 @@ installed produces the same bytes as before, as do functions encoded by name.
 Regenerate any plan you serialized with an earlier release and stored for later
 use, if it was produced by a session with an extension codec installed.
 
+### Physical plans report their partitioning scheme
+
+{py:attr}`~datafusion.ExecutionPlan.output_partitioning` is new, and reports
+what {py:attr}`~datafusion.ExecutionPlan.partition_count` leaves out: whether
+the rows in those partitions are hash-distributed on known keys, spread
+round-robin, or merely counted. It returns a
+{py:class}`~datafusion.PhysicalPartitioning`.
+
+```python
+partitioning = df.execution_plan().output_partitioning
+partitioning.scheme  # 'Hash', 'RoundRobinBatch', 'Range', 'UnknownPartitioning'
+partitioning.partition_count
+partitioning.hash_expressions  # display strings, or None
+```
+
+This is additive; `partition_count` keeps working and agrees with
+`output_partitioning.partition_count`. Note that
+{py:class}`datafusion.expr.Partitioning` is a different type: that one is the
+*logical* partitioning {py:meth}`~datafusion.DataFrame.repartition_by_hash`
+takes as a request, while this one is what a built plan actually does.
+
+### Two error paths that used to abort the interpreter
+
+{py:meth}`~datafusion.SessionContext.execute` now raises `ValueError` for a
+partition index that is out of range. Previously the plan's leaves indexed
+their partitions directly and the resulting Rust panic surfaced as an error
+naming neither the plan nor the index.
+
+{py:meth}`~datafusion.SessionConfig.set` now raises for a key whose namespace
+does not exist -- `datafusion.runtime.*`, or a config extension that has not
+been installed yet. Previously it aborted with a `PanicException`, which
+derives from `BaseException` and so escaped `except Exception`. This matters
+when replaying settings read back from `information_schema.df_settings`, which
+lists keys in both of those categories.
+
 ### Changes to the `datafusion-python-util` crate
 
 Extension libraries written in Rust usually depend on the
 
@@ -122,9 +122,24 @@ impl PhysicalExtensionCodec for CountingPhysicalExtensionCodec {
         buf: &mut Vec<u8>,
         proto_converter: &dyn PhysicalProtoConverterExtension,
     ) -> Result<()> {
-        // The provider owns DataSourceExec. A ForeignExecutionPlan can wrap a
-        // host-added execution decorator around that scan; retaining the opaque
-        // wrapper preserves its original library identity without downcasting it.
+        // `DataSourceExec` is this library's own node. The `ForeignExecutionPlan`
+        // arm is a workaround, not a pattern to copy, and it is load-bearing:
+        // a host physical optimizer rule that runs during a foreign planner's
+        // `create_physical_plan` -- `EnsureCooperative` always does -- hands the
+        // library back a `ForeignExecutionPlan` wrapping the host's
+        // `CooperativeExec`. That type has no reachable `try_to_proto`, so
+        // nothing can encode it natively and `FFI_QueryPlanner` must serialize
+        // the plan it returns. Claiming it here is what lets those plans
+        // round-trip at all.
+        //
+        // The cost is that this codec also claims every *other* library's
+        // nodes, since that is the type any node arrives as once it has crossed
+        // the boundary -- see `extension_codec_order`. Narrowing this to
+        // `DataSourceExec` alone makes 31 tests in
+        // `datafusion-ffi-query-planner-example` fail with the error above.
+        //
+        // A library whose planner controls its own physical optimizer rules
+        // never sees a foreign node and needs no such arm.
         if node.is::<DataSourceExec>() || node.is::<ForeignExecutionPlan>() {
             self.counters
                 .encode_execution_plan
 
@@ -100,7 +100,13 @@
 )
 from .io import read_avro, read_csv, read_json, read_parquet
 from .options import CsvReadOptions
-from .plan import ExecutionPlan, LogicalPlan, Metric, MetricsSet
+from .plan import (
+    ExecutionPlan,
+    LogicalPlan,
+    Metric,
+    MetricsSet,
+    PhysicalPartitioning,
+)
 from .record_batch import RecordBatch, RecordBatchStream
 from .user_defined import (
     Accumulator,
@@ -133,6 +139,7 @@
     "MetricsSet",
     "ParquetColumnOptions",
     "ParquetWriterOptions",
+    "PhysicalPartitioning",
     "QueryPlannerExportable",
     "RecordBatch",
     "RecordBatchStream",