refactor(ffi-example): MyExecutionPlan emits real data via

MemorySourceConfig

Was a one-column `EmptyExec` stub useful only as a capsule-handoff
target. Promoted to a minimal reference impl that a downstream Rust
crate can copy when exposing a custom `ExecutionPlan` to
datafusion-python: configurable `num_rows`, produces a single batch
of sequential `Int32` values under column `value`, wrapped in
`DataSourceExec` via `MemorySourceConfig::try_new_exec`. Header
comment explains the typical use case (remote backend, streaming
source, synthetic data generator) and the
`__datafusion_execution_plan__` capsule shape downstream crates
should follow.

Test asserts the schema-bearing plan survives the FFI hop: a
`DataSourceExec` arrives with the expected partitioning and no
children. Schema details are not surfaced through the FFI display
path (only the wrapping `ForeignExecutionPlan` name + inner plan
name appear), so the test does not assert the column name.

`to_bytes` round-trip of an FFI-imported plan is not exercised:
encoding requires a physical codec that knows how to serialize
`ForeignExecutionPlan`, which the default codec does not. A
downstream user round-tripping such a plan must install their own
codec via `with_physical_extension_codec`. Documented in the test
file rather than asserted on.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: timsaucer

examples/datafusion-ffi-example/python/tests/_test_execution_plan.py

@@ -24,16 +24,31 @@
 
 @pytest.mark.parametrize("inner_capsule", [True, False])
 def test_execution_plan_from_pycapsule(inner_capsule: bool) -> None:
-    """`ExecutionPlan.from_pycapsule` consumes a capsule sourced from a
-    non-datafusion-python crate, regardless of whether the user passes
-    the typed wrapper or pre-extracts the raw capsule."""
-    obj = MyExecutionPlan()
+    """`ExecutionPlan.from_pycapsule` consumes a capsule sourced from
+    a downstream Rust crate, regardless of whether the user passes the
+    typed wrapper or pre-extracts the raw capsule.
+
+    `MyExecutionPlan` produces a real `DataSourceExec` backed by a
+    `MemorySourceConfig`. After the FFI handoff the underlying plan
+    is wrapped in a `ForeignExecutionPlan`, so the display surface
+    shows the FFI envelope plus the wrapped plan's name — schema
+    detail is not propagated by the FFI display path.
+    """
+    obj = MyExecutionPlan(num_rows=5)
     if inner_capsule:
         obj = obj.__datafusion_execution_plan__()
 
     plan = ExecutionPlan.from_pycapsule(obj)
 
-    # EmptyExec round-trips with no children.
     assert plan.children() == []
     assert plan.partition_count == 1
-    assert "EmptyExec" in plan.display()
+    assert "DataSourceExec" in plan.display()
+
+
+# `to_bytes` round-trip of an FFI-imported `ExecutionPlan` requires a
+# physical codec that knows how to encode `ForeignExecutionPlan`. The
+# default codec does not, so a downstream user serializing such a plan
+# must install their own physical extension codec via
+# `SessionContext.with_physical_extension_codec(...)`. We do not
+# exercise that here because the test would assert the encode path on
+# the user's codec, not on datafusion-python's plumbing.

examples/datafusion-ffi-example/src/execution_plan.rs

@@ -15,42 +15,86 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//! Reference implementation of a downstream-crate `ExecutionPlan`
+//! exported to datafusion-python via the PyCapsule protocol.
+//!
+//! Typical use case: a Rust crate produces a custom physical plan
+//! (e.g. for a remote backend, a streaming source, a synthetic data
+//! generator) and wants to hand it to a datafusion-python user
+//! without exposing its internals. The downstream crate writes a
+//! `#[pyclass]` wrapper, builds an `Arc<dyn ExecutionPlan>` on
+//! demand, wraps it as `FFI_ExecutionPlan`, and exposes the capsule
+//! via `__datafusion_execution_plan__(...)`. The receiving Python
+//! code calls
+//! `datafusion.ExecutionPlan.from_pycapsule(my_plan)` and gets back
+//! a fully-typed datafusion-python `ExecutionPlan` it can introspect,
+//! serialize via `to_bytes` / `from_bytes`, or execute.
+//!
+//! `MyExecutionPlan` here is a minimal worked example: it produces
+//! `num_rows` sequential `Int32` values in a single batch under the
+//! column name `value`. Backed by `MemorySourceConfig`, so the plan
+//! is a real `DataSourceExec` with a non-trivial schema and
+//! executable output — useful both as a reference for downstream
+//! crates and as the fixture for the FFI integration tests.
+
 use std::sync::Arc;
 
+use arrow_array::{Int32Array, RecordBatch};
 use arrow_schema::{DataType, Field, Schema};
-use datafusion::physical_plan::empty::EmptyExec;
+use datafusion::datasource::memory::MemorySourceConfig;
+use datafusion::physical_plan::ExecutionPlan;
+use datafusion_common::error::{DataFusionError, Result as DataFusionResult};
 use datafusion_ffi::execution_plan::FFI_ExecutionPlan;
 use datafusion_python_util::get_tokio_runtime;
+use pyo3::exceptions::PyRuntimeError;
 use pyo3::prelude::*;
 use pyo3::types::PyCapsule;
 
-/// Minimal downstream `ExecutionPlan` producer for FFI integration
-/// tests. Emits a single-column empty plan via `FFI_ExecutionPlan`,
-/// letting Python tests verify that
-/// `ExecutionPlan.from_pycapsule(my_plan)` consumes a capsule sourced
-/// from a non-datafusion-python Rust crate.
+/// Downstream-crate `ExecutionPlan` producer for FFI integration with
+/// datafusion-python. Emits `num_rows` sequential `Int32` values
+/// (0..num_rows) in a single batch under the column name `value`.
 #[pyclass(
     from_py_object,
     name = "MyExecutionPlan",
     module = "datafusion_ffi_example",
     subclass
 )]
 #[derive(Clone)]
-pub(crate) struct MyExecutionPlan;
+pub(crate) struct MyExecutionPlan {
+    num_rows: usize,
+}
+
+impl MyExecutionPlan {
+    fn build_plan(&self) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "value",
+            DataType::Int32,
+            false,
+        )]));
+        let values: Vec<i32> = (0..self.num_rows as i32).collect();
+        let batch = RecordBatch::try_new(
+            Arc::clone(&schema),
+            vec![Arc::new(Int32Array::from(values))],
+        )?;
+        let exec = MemorySourceConfig::try_new_exec(&[vec![batch]], schema, None)?;
+        Ok(exec as Arc<dyn ExecutionPlan>)
+    }
+}
 
 #[pymethods]
 impl MyExecutionPlan {
     #[new]
-    fn new() -> Self {
-        Self
+    fn new(num_rows: usize) -> Self {
+        Self { num_rows }
     }
 
     fn __datafusion_execution_plan__<'py>(
         &self,
         py: Python<'py>,
     ) -> PyResult<Bound<'py, PyCapsule>> {
-        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)]));
-        let plan = Arc::new(EmptyExec::new(schema));
+        let plan = self
+            .build_plan()
+            .map_err(|e: DataFusionError| PyRuntimeError::new_err(e.to_string()))?;
         let runtime = get_tokio_runtime().handle().clone();
         let ffi = FFI_ExecutionPlan::new(plan, Some(runtime));