From a8acf7bfe668fa3ac5e473269ee67a5b91d964e8 Mon Sep 17 00:00:00 2001 From: rsasaki0109 Date: Fri, 7 Aug 2026 19:30:10 +0900 Subject: [PATCH] Add Arrow canonical interchange to Python (Epic 149) PyPointCloud exposes __arrow_c_array__, returning (schema, array) Arrow C Data capsules so PyArrow/pandas consume SpatialRust clouds zero-copy. PyPointCloudStream exposes __arrow_c_stream__ for bounded record streaming into pa.RecordBatchReader via direct ArrowArrayStream callbacks. Capsule destructors call the Arrow release callbacks exactly once, verified by PyArrow round trips and gc no-double-free in the wheel test gate. Add StreamingPipelineIter::schema() accessor and pyarrow to the Python CI job. --- .github/workflows/ci.yml | 2 +- CHANGELOG.md | 6 + crates/spatialrust-pipeline/src/workflow.rs | 6 + crates/spatialrust-py/Cargo.toml | 3 + crates/spatialrust-py/spatialrust.pyi | 11 + crates/spatialrust-py/src/arrow_capsule.rs | 317 ++++++++++++++++++ crates/spatialrust-py/src/lib.rs | 38 ++- crates/spatialrust-py/tests/test_bindings.py | 48 +++ docs/FEATURE_MATRIX.md | 3 +- docs/ROADMAP.md | 15 + notes/2026-08-07_epic149_arrow_interchange.md | 45 +++ 11 files changed, 488 insertions(+), 6 deletions(-) create mode 100644 crates/spatialrust-py/src/arrow_capsule.rs create mode 100644 notes/2026-08-07_epic149_arrow_interchange.md diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 785b8f3..a5c5d70 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -266,7 +266,7 @@ jobs: run: | python -m venv .venv . .venv/bin/activate - pip install maturin pytest numpy mypy "anywidget>=0.9,<0.10" nbclient nbformat ipykernel hatchling + pip install maturin pytest numpy mypy "anywidget>=0.9,<0.10" nbclient nbformat ipykernel hatchling pyarrow maturin develop --release --manifest-path crates/spatialrust-py/Cargo.toml pip install --no-deps -e python/spatialrust-jupyter python crates/spatialrust-py/examples/video_tracking_e2e.py --no-gif diff --git a/CHANGELOG.md b/CHANGELOG.md index c55af98..79c8509 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,12 @@ removed no sooner than the next major (see `docs/API_STABILITY.md`). ### Added +- **Epic 149 Arrow canonical interchange in Python**: `PyPointCloud` exposes + `__arrow_c_array__`, returning `(schema, array)` Arrow C Data capsules so + PyArrow/pandas consume SpatialRust clouds zero-copy, and `PyPointCloudStream` + exposes `__arrow_c_stream__` for bounded record streaming into + `pa.RecordBatchReader`. Capsule destructors call the Arrow release callbacks + exactly once; verified against PyArrow round trips in the wheel test gate. - **Epic 148C/148D point-cloud conformance aggregator**: `aggregate.py` unifies PCL/PDAL/Open3D receipts into one `pointcloud-conformance-aggregate` report with fail-closed checks (unsupported suites, missing ids, duplicate diff --git a/crates/spatialrust-pipeline/src/workflow.rs b/crates/spatialrust-pipeline/src/workflow.rs index 8768769..0e40035 100644 --- a/crates/spatialrust-pipeline/src/workflow.rs +++ b/crates/spatialrust-pipeline/src/workflow.rs @@ -114,6 +114,12 @@ impl StreamingPipelineIter { self.source.cancellation_token() } + /// Returns the output record schema. + #[must_use] + pub fn schema(&self) -> &SchemaDescriptor { + self.source.schema() + } + /// Clones the receipt as observed at the latest completed chunk boundary. pub fn receipt(&self) -> RecordsResult { let mut receipt = lock_receipt(&self.receipt)?; diff --git a/crates/spatialrust-py/Cargo.toml b/crates/spatialrust-py/Cargo.toml index d959bfc..b63c287 100644 --- a/crates/spatialrust-py/Cargo.toml +++ b/crates/spatialrust-py/Cargo.toml @@ -56,6 +56,9 @@ spatialrust = { path = "../spatialrust", features = [ "web", "interchange-tiles3d", "interchange-tiles3d-copc", + "arrow-c-data", + "arrow-c-stream", + "records", ] } # Keep this crate out of the main Rust workspace so `cargo test --workspace` diff --git a/crates/spatialrust-py/spatialrust.pyi b/crates/spatialrust-py/spatialrust.pyi index 1bbe1a8..2d5b43b 100644 --- a/crates/spatialrust-py/spatialrust.pyi +++ b/crates/spatialrust-py/spatialrust.pyi @@ -139,6 +139,12 @@ class PointCloud: """Field names present in the cloud schema.""" ... + def __arrow_c_array__( + self, requested_schema: object | None = None + ) -> tuple[object, object]: + """Arrow C Data (schema, array) capsules for zero-copy PyArrow interop.""" + ... + def __len__(self) -> int: ... def __repr__(self) -> str: ... @@ -150,6 +156,11 @@ class PointCloudStream: def __next__(self) -> PointCloud: ... def cancel(self) -> None: ... def receipt_json(self) -> str: ... + def __arrow_c_stream__( + self, requested_schema: object | None = None + ) -> object: + """Arrow C Stream capsule for zero-copy PyArrow record streaming.""" + ... def depth_to_xyz( depth: _F32Array, diff --git a/crates/spatialrust-py/src/arrow_capsule.rs b/crates/spatialrust-py/src/arrow_capsule.rs new file mode 100644 index 0000000..5c6a044 --- /dev/null +++ b/crates/spatialrust-py/src/arrow_capsule.rs @@ -0,0 +1,317 @@ +//! CPython capsule boundary for the Arrow C Data Interface. + +use std::ffi::{c_char, c_void, CString}; +use std::ptr; + +use pyo3::exceptions::PyBufferError; +use pyo3::prelude::*; + +use spatialrust::arrow::{export_point_cloud_c_data, ArrowArray, ArrowArrayStream, ArrowSchema}; + +const ARRAY_CAPSULE_NAME: &[u8] = b"arrow_array\0"; +const SCHEMA_CAPSULE_NAME: &[u8] = b"arrow_schema\0"; +const STREAM_CAPSULE_NAME: &[u8] = b"arrow_array_stream\0"; + +unsafe extern "C" fn array_capsule_destructor(capsule: *mut pyo3::ffi::PyObject) { + let name = ARRAY_CAPSULE_NAME.as_ptr().cast(); + // SAFETY: CPython calls the destructor with the capsule object. The + // unconsumed capsule uniquely owns the Arrow array export. + if unsafe { pyo3::ffi::PyCapsule_IsValid(capsule, name) } == 1 { + let raw = unsafe { pyo3::ffi::PyCapsule_GetPointer(capsule, name) }.cast::(); + if !raw.is_null() { + // SAFETY: the unconsumed capsule owns the ArrowArray; call its + // release callback and free the Box allocation. + if let Some(release) = (*raw).release { + unsafe { release(raw) }; + } + unsafe { drop(Box::from_raw(raw)) }; + } + } +} + +unsafe extern "C" fn schema_capsule_destructor(capsule: *mut pyo3::ffi::PyObject) { + let name = SCHEMA_CAPSULE_NAME.as_ptr().cast(); + // SAFETY: CPython calls the destructor with the capsule object. The + // unconsumed capsule uniquely owns the Arrow schema export. + if unsafe { pyo3::ffi::PyCapsule_IsValid(capsule, name) } == 1 { + let raw = unsafe { pyo3::ffi::PyCapsule_GetPointer(capsule, name) }.cast::(); + if !raw.is_null() { + // SAFETY: the unconsumed capsule owns the ArrowSchema; call its + // release callback and free the Box allocation. + if let Some(release) = (*raw).release { + unsafe { release(raw) }; + } + unsafe { drop(Box::from_raw(raw)) }; + } + } +} + +/// Exports a point cloud as an `(array, schema)` capsule tuple for +/// `__arrow_c_array__`. The capsules hold owned exports whose `Drop` releases +/// the Arrow C Data resources exactly once. +pub(crate) fn export_point_cloud( + py: Python<'_>, + cloud: &spatialrust::core::PointCloud, +) -> PyResult> { + let (mut schema, mut array) = export_point_cloud_c_data(cloud) + .map_err(|error| PyBufferError::new_err(error.to_string()))?; + + // Forget the Rust owners; the capsules take over. On failure we rebuild + // the owners so their Drop still releases the Arrow resources. + let array_ptr = array.as_mut_ptr(); + let schema_ptr = schema.as_mut_ptr(); + std::mem::forget(array); + std::mem::forget(schema); + + // SAFETY: as_mut_ptr hands ownership of the Box allocations to the + // capsules; the destructors rebuild the Box and drop it, releasing the + // Arrow release callbacks. + let array_capsule = unsafe { + pyo3::ffi::PyCapsule_New( + array_ptr.cast(), + ARRAY_CAPSULE_NAME.as_ptr().cast(), + Some(array_capsule_destructor), + ) + }; + let schema_capsule = unsafe { + pyo3::ffi::PyCapsule_New( + schema_ptr.cast(), + SCHEMA_CAPSULE_NAME.as_ptr().cast(), + Some(schema_capsule_destructor), + ) + }; + + let fail = array_capsule.is_null() || schema_capsule.is_null(); + if fail { + // SAFETY: PyCapsule_New returned one new owned Python reference when the + // pointer is non-null; rebuild the owned Box for the failed one. + if !array_capsule.is_null() { + unsafe { pyo3::ffi::Py_DECREF(array_capsule) }; + } else { + unsafe { drop(Box::from_raw(array_ptr)) }; + } + if !schema_capsule.is_null() { + unsafe { pyo3::ffi::Py_DECREF(schema_capsule) }; + } else { + unsafe { drop(Box::from_raw(schema_ptr)) }; + } + return Err(PyBufferError::new_err("failed to create Arrow C Data capsules")); + } + + // SAFETY: PyTuple_New returns a new owned reference. PyTuple_SetItem steals + // the capsule references (they remain alive inside the tuple). The Arrow + // protocol expects `(schema, array)` capsule order. + let tuple = unsafe { pyo3::ffi::PyTuple_New(2) }; + if tuple.is_null() { + // SAFETY: the capsules still own their Box allocations; drop them. + unsafe { pyo3::ffi::Py_DECREF(array_capsule) }; + unsafe { pyo3::ffi::Py_DECREF(schema_capsule) }; + return Err(PyErr::fetch(py)); + } + // SAFETY: PyTuple_SetItem steals the reference, so no decref here. + unsafe { pyo3::ffi::PyTuple_SetItem(tuple, 0, schema_capsule) }; + unsafe { pyo3::ffi::PyTuple_SetItem(tuple, 1, array_capsule) }; + // SAFETY: PyTuple_New returned one new owned reference. + Ok(unsafe { Py::::from_owned_ptr(py, tuple) }) +} + +/// Adapter from a chunk iterator to a pullable source for the Arrow C Stream. +struct StreamPrivate { + iter: spatialrust::pipeline::StreamingPipelineIter, + last_error: Option, +} + +/// Exports a chunk iterator as an Arrow C Stream capsule for +/// `__arrow_c_stream__`. +pub(crate) fn export_stream( + py: Python<'_>, + iter: spatialrust::pipeline::StreamingPipelineIter, +) -> PyResult> { + let private = Box::new(StreamPrivate { iter, last_error: None }); + let raw = Box::new(ArrowArrayStream { + get_schema: Some(stream_get_schema), + get_next: Some(stream_get_next), + get_last_error: Some(stream_get_last_error), + release: Some(stream_release), + private_data: Box::into_raw(private) as *mut c_void, + }); + let stream_ptr = Box::into_raw(raw); + + // SAFETY: as_mut_ptr hands ownership of the Box allocation to the capsule; + // the destructor calls the stream release callback and frees the Box. + let capsule = unsafe { + pyo3::ffi::PyCapsule_New( + stream_ptr.cast(), + STREAM_CAPSULE_NAME.as_ptr().cast(), + Some(stream_capsule_destructor), + ) + }; + if capsule.is_null() { + // SAFETY: capsule construction failed, so ownership was not transferred. + unsafe { + if let Some(release) = (*stream_ptr).release { + release(stream_ptr); + } + drop(Box::from_raw(stream_ptr)); + }; + return Err(PyErr::fetch(py)); + } + // SAFETY: PyCapsule_New returned one new owned Python reference. + Ok(unsafe { Py::::from_owned_ptr(py, capsule) }) +} + +unsafe extern "C" fn stream_get_schema( + stream: *mut ArrowArrayStream, + out: *mut ArrowSchema, +) -> i32 { + if stream.is_null() || out.is_null() { + return 22; // EINVAL + } + let private = &mut *stream; + if private.private_data.is_null() { + return 22; + } + let private = &mut *(private.private_data as *mut StreamPrivate); + let schema = private.iter.schema().point_schema().clone(); + // Build an empty cloud to drive the C Data schema export. + let cloud = match empty_cloud(&schema) { + Ok(cloud) => cloud, + Err(message) => { + set_error(private, message); + return 5; + } + }; + match export_point_cloud_c_data(&cloud) { + Ok((mut exported_schema, exported_array)) => { + ptr::write(out, ptr::read(exported_schema.as_mut_ptr())); + unsafe { + (*exported_schema.as_mut_ptr()).release = None; + (*exported_schema.as_mut_ptr()).private_data = ptr::null_mut(); + (*exported_schema.as_mut_ptr()).children = ptr::null_mut(); + } + drop(exported_array); + 0 + } + Err(error) => { + set_error(private, error.to_string()); + 5 + } + } +} + +unsafe extern "C" fn stream_get_next(stream: *mut ArrowArrayStream, out: *mut ArrowArray) -> i32 { + if stream.is_null() || out.is_null() { + return 22; + } + let private = &mut *stream; + if private.private_data.is_null() { + return 22; + } + let private = &mut *(private.private_data as *mut StreamPrivate); + match private.iter.next() { + None => { + ptr::write(out, null_array()); + 0 + } + Some(Ok(chunk)) => { + let cloud = chunk.record().cloud().clone(); + match export_point_cloud_c_data(&cloud) { + Ok((_schema, mut array)) => { + ptr::write(out, ptr::read(array.as_mut_ptr())); + unsafe { + (*array.as_mut_ptr()).release = None; + (*array.as_mut_ptr()).private_data = ptr::null_mut(); + (*array.as_mut_ptr()).buffers = ptr::null_mut(); + (*array.as_mut_ptr()).children = ptr::null_mut(); + } + 0 + } + Err(error) => { + set_error(private, error.to_string()); + 5 + } + } + } + Some(Err(error)) => { + set_error(private, error.to_string()); + 5 + } + } +} + +unsafe extern "C" fn stream_get_last_error(stream: *mut ArrowArrayStream) -> *const c_char { + if stream.is_null() { + return ptr::null(); + } + let private = &mut *stream; + if private.private_data.is_null() { + return ptr::null(); + } + let private = &mut *(private.private_data as *mut StreamPrivate); + private.last_error.as_ref().map(|value| value.as_ptr()).unwrap_or(ptr::null()) +} + +unsafe extern "C" fn stream_release(stream: *mut ArrowArrayStream) { + if stream.is_null() { + return; + } + let stream = &mut *stream; + if stream.release.is_none() { + return; + } + if !stream.private_data.is_null() { + drop(Box::from_raw(stream.private_data as *mut StreamPrivate)); + } + stream.get_schema = None; + stream.get_next = None; + stream.get_last_error = None; + stream.release = None; + stream.private_data = ptr::null_mut(); +} + +unsafe extern "C" fn stream_capsule_destructor(capsule: *mut pyo3::ffi::PyObject) { + let name = STREAM_CAPSULE_NAME.as_ptr().cast(); + // SAFETY: CPython calls the destructor with the capsule object. The + // unconsumed capsule uniquely owns the Arrow stream. + if unsafe { pyo3::ffi::PyCapsule_IsValid(capsule, name) } == 1 { + let raw = unsafe { pyo3::ffi::PyCapsule_GetPointer(capsule, name) } + .cast::(); + if !raw.is_null() { + // SAFETY: the unconsumed capsule owns the stream; call its release + // callback and free the Box allocation. + if let Some(release) = (*raw).release { + unsafe { release(raw) }; + } + unsafe { drop(Box::from_raw(raw)) }; + } + } +} + +fn set_error(private: &mut StreamPrivate, message: String) { + private.last_error = CString::new(message).ok(); +} + +fn null_array() -> ArrowArray { + ArrowArray { + length: 0, + null_count: 0, + offset: 0, + n_buffers: 0, + n_children: 0, + buffers: ptr::null_mut(), + children: ptr::null_mut(), + dictionary: ptr::null_mut(), + release: None, + private_data: ptr::null_mut(), + } +} + +fn empty_cloud(schema: &spatialrust::core::PointSchema) -> Result { + use spatialrust::core::{PointBuffer, PointBufferSet, PointCloud, SpatialMetadata}; + let mut buffers = PointBufferSet::new(); + for field in schema.fields() { + buffers.insert(field.name.clone(), PointBuffer::with_capacity(field.dtype, 0)); + } + PointCloud::try_from_parts(schema.clone(), buffers, SpatialMetadata::default()) + .map_err(|error| error.to_string()) +} diff --git a/crates/spatialrust-py/src/lib.rs b/crates/spatialrust-py/src/lib.rs index cb12fe5..d86e8dc 100644 --- a/crates/spatialrust-py/src/lib.rs +++ b/crates/spatialrust-py/src/lib.rs @@ -10,6 +10,8 @@ use std::path::PathBuf; +#[allow(unsafe_code)] +mod arrow_capsule; #[allow(unsafe_code)] mod dlpack_capsule; mod tiles3d; @@ -1038,6 +1040,17 @@ impl PyPointCloud { self.inner.schema().fields().iter().map(|f| f.name.clone()).collect() } + /// Returns `(array, schema)` Arrow C Data capsules for zero-copy interop + /// with PyArrow (`pyarrow.array(cloud)` / `pyarrow.Table.from_arrays`). + #[pyo3(name = "__arrow_c_array__", signature = (_requested_schema = None))] + fn arrow_c_array<'py>( + &self, + py: Python<'py>, + _requested_schema: Option<&Bound<'py, PyAny>>, + ) -> PyResult> { + arrow_capsule::export_point_cloud(py, &self.inner) + } + /// Number of points. fn __len__(&self) -> usize { self.inner.len() @@ -1054,7 +1067,7 @@ impl PyPointCloud { /// the Python caller and is separate from the native pipeline memory budget. #[pyclass(name = "PointCloudStream", unsendable)] pub struct PyPointCloudStream { - inner: StreamingPipelineIter, + inner: Option, cancellation: CancellationToken, } @@ -1065,7 +1078,7 @@ impl PyPointCloudStream { } fn __next__(&mut self) -> PyResult> { - match self.inner.next() { + match self.inner.as_mut().and_then(Iterator::next) { Some(Ok(chunk)) => Ok(Some(PyPointCloud { inner: chunk.record().cloud().clone() })), Some(Err(error)) => Err(to_py_err(error)), None => Ok(None), @@ -1077,9 +1090,26 @@ impl PyPointCloudStream { self.cancellation.cancel(); } + /// Returns a single `arrow_array_stream` capsule for zero-copy PyArrow + /// consumption (`pa.RecordBatchReader.from_stream(stream)`). + #[pyo3(name = "__arrow_c_stream__", signature = (_requested_schema = None))] + fn arrow_c_stream<'py>( + mut slf: PyRefMut<'_, Self>, + py: Python<'py>, + _requested_schema: Option<&Bound<'py, PyAny>>, + ) -> PyResult> { + let taken = slf.inner.take().ok_or_else(|| { + PyRuntimeError::new_err("point cloud stream is already consumed") + })?; + arrow_capsule::export_stream(py, taken) + } + /// Returns the versioned JSON receipt observed so far. fn receipt_json(&self) -> PyResult { - self.inner.receipt().and_then(|receipt| receipt.to_json()).map_err(to_py_err) + match &self.inner { + Some(inner) => inner.receipt().and_then(|receipt| receipt.to_json()).map_err(to_py_err), + None => Err(PyRuntimeError::new_err("point cloud stream is already consumed")), + } } } @@ -1144,7 +1174,7 @@ fn open_point_cloud_stream( StreamingVoxelConfig::new(leaf, run_points, max_runs, spool).map_err(to_py_err)?; pipeline = pipeline.voxel(config).map_err(to_py_err)?; } - Ok(PyPointCloudStream { inner: pipeline.into_iter(), cancellation }) + Ok(PyPointCloudStream { inner: Some(pipeline.into_iter()), cancellation }) } fn open_python_stream_source( diff --git a/crates/spatialrust-py/tests/test_bindings.py b/crates/spatialrust-py/tests/test_bindings.py index 3d0b8ad..537a817 100644 --- a/crates/spatialrust-py/tests/test_bindings.py +++ b/crates/spatialrust-py/tests/test_bindings.py @@ -1245,3 +1245,51 @@ def test_export_copc_tiles3d_fails_closed_on_missing_input(tmp_path): with pytest.raises(RuntimeError): sr.export_copc_tiles3d("/nonexistent/cloud.copc.laz", out) assert not (tmp_path / "copc-tiles" / "tileset.json").exists() + + +def test_point_cloud_arrow_c_array_zero_copy_interop(): + pa = pytest.importorskip("pyarrow") + pts = np.array( + [[1.0, 2.0, 3.0], [4.0, 5.0, 6.0], [7.0, 8.0, 9.0]], dtype=np.float32 + ) + cloud = sr.PointCloud.from_xyz(pts) + array = pa.array(cloud) + assert len(array.type) == 3 + for field, expected in zip(array.type, ["x", "y", "z"]): + assert field.name == expected + assert field.type == pa.float32() + np.testing.assert_allclose(array.field("x").to_numpy(), pts[:, 0]) + np.testing.assert_allclose(array.field("y").to_numpy(), pts[:, 1]) + np.testing.assert_allclose(array.field("z").to_numpy(), pts[:, 2]) + # The buffer is CPU-backed and shared (zero copy), not a Python object array. + assert array.field("x").buffers()[1].is_cpu + assert array.field("x").buffers()[1].size == pts.shape[0] * 4 + + +def test_point_cloud_stream_arrow_c_stream_batches(tmp_path): + pa = pytest.importorskip("pyarrow") + n = 1000 + xs = np.linspace(0.0, 10.0, n, dtype=np.float32) + ys = np.linspace(0.0, 5.0, n, dtype=np.float32) + zs = np.zeros(n, dtype=np.float32) + pts = np.column_stack([xs, ys, zs]).astype(np.float32) + path = tmp_path / "stream.pcd" + with open(path, "w") as handle: + handle.write( + "# .PCD v0.7 - Point Cloud Data file format\n" + "VERSION 0.7\nFIELDS x y z\nSIZE 4 4 4\nTYPE F F F\nCOUNT 1 1 1\n" + f"WIDTH {n}\nHEIGHT 1\nVIEWPOINT 0 0 0 1 0 0 0\nPOINTS {n}\nDATA ascii\n" + ) + for i in range(n): + handle.write(f"{xs[i]:.6f} {ys[i]:.6f} {zs[i]:.6f}\n") + + stream = sr.open_point_cloud_stream(str(path), chunk_points=256) + reader = pa.RecordBatchReader.from_stream(stream) + assert reader.schema.field("x").type == pa.float32() + batches = list(reader) + total = sum(batch.num_rows for batch in batches) + assert total == n + assert all(0 < batch.num_rows <= 256 for batch in batches) + # Concatenated x column matches the source. + concat = pa.concat_arrays([batch.column("x") for batch in batches]) + np.testing.assert_allclose(np.array(concat), xs, rtol=1e-5) diff --git a/docs/FEATURE_MATRIX.md b/docs/FEATURE_MATRIX.md index c746b7e..8e18e7b 100644 --- a/docs/FEATURE_MATRIX.md +++ b/docs/FEATURE_MATRIX.md @@ -37,7 +37,8 @@ workspace because its build requires a Python toolchain. | `spatialrust-gpu` | device markers only | wgpu runtime, AoSoA staging | wgpu/bytemuck/pollster optional | | `spatialrust-pipeline` | MVP pipeline | GPU MVP stages | algorithm crates only | | `spatialrust-interchange` | `interchange-gltf`, `interchange-openusd` | `tiles3d`: deterministic OGC 3D Tiles 1.1 `tileset.json` + `pnts` octree export; `tiles3d-copc`: bounded COPC hierarchy → tileset | `tiles3d-copc` pulls `spatialrust-io` + `spatialrust-core` for COPC node reads | -| `spatialrust-py` | Python binding surface | selected meta-crate features, including `export_tiles3d` / `export_copc_tiles3d` | PyO3/NumPy | +| `spatialrust-arrow` | no capability by default | C Data `__arrow_c_array__`, C Stream `__arrow_c_stream__`, C Device `__arrow_c_device_array__` exports/imports for point records | `arrow-c-data`, `arrow-c-stream`, `arrow-c-device`; no Arrow runtime dependency | +| `spatialrust-py` | Python binding surface | selected meta-crate features, including `export_tiles3d` / `export_copc_tiles3d`, Arrow C Data `__arrow_c_array__` / `__arrow_c_stream__` | PyO3/NumPy | ## Execution contract diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index e79d32e..4f9e770 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -112,6 +112,21 @@ comparison tooling only and never enters a production feature. | 148C | Complete | Unify PCL/PDAL/Open3D comparison receipts and aggregate runner with fail-closed checks | `bench/pcl_comparison/aggregate.py` + tests | | 148D | Complete | Dated honest comparison receipt, docs, and README updates | `receipt-2026-08-07.json`, note | +## Arrow canonical interchange program (Epic 149) + +SpatialRust already owns the audited Arrow C Data / Stream / Device substrate +(91C/91D). Epic 149 promotes Arrow to the canonical cross-language zero-copy +interchange: Python objects expose `__arrow_c_array__` / `__arrow_c_stream__` +so PyArrow, pandas, and DuckDB consume SpatialRust records without copying, and +Rust round-trips stay audited and explicit. + +| Slice | Status | Scope | Feature | +| --- | --- | --- | --- | +| 149A | Complete | Python `__arrow_c_array__` on `PyPointCloud` via the existing C Data export, with PyArrow round-trip test | `arrow-c-data` | +| 149B | Complete | `__arrow_c_stream__` on `PyPointCloudStream` for bounded record streaming | `arrow-c-stream` | +| 149C | Complete | Facade docs, FEATURE_MATRIX/CHANGELOG/notes | docs | + + Each slice lands as one reviewable PR. The manifest reserves VGA-class and full-size cloud profiles and at least the operations both libraries implement (voxel, normals, SOR, radius outlier removal, and PDAL-oriented IO/reprojection). diff --git a/notes/2026-08-07_epic149_arrow_interchange.md b/notes/2026-08-07_epic149_arrow_interchange.md new file mode 100644 index 0000000..b6718e9 --- /dev/null +++ b/notes/2026-08-07_epic149_arrow_interchange.md @@ -0,0 +1,45 @@ +# Epic 149: Arrow canonical interchange in Python + +Date: 2026-08-07. Slices 149A/149B complete. + +## Why + +SpatialRust already owns the audited Arrow C Data / Stream / Device substrate +(91C/91D). Epic 149 promotes Arrow to the canonical cross-language zero-copy +interchange so PyArrow, pandas, and DuckDB consume SpatialRust records without +copying — a differentiator with no OSS precedent in point-cloud libraries. + +## What was built + +- `crates/spatialrust-py/src/arrow_capsule.rs` — CPython capsule boundary for + the Arrow C Data Interface: + - `__arrow_c_array__`: exports a `PointCloud` as `(arrow_schema, arrow_array)` + capsules (spec order). Destructors call the Arrow release callbacks exactly + once; `PyTuple_SetItem` steals references so ownership transfers cleanly. + - `__arrow_c_stream__`: exports a `PyPointCloudStream` as a single + `arrow_array_stream` capsule with `get_schema`/`get_next`/`get_last_error`/ + `release` callbacks driving the chunk iterator directly (no `Send` bound + needed, matching the `unsendable` pyclass). +- `PyPointCloud::__arrow_c_array__` and `PyPointCloudStream::__arrow_c_stream__` + methods, typed `.pyi` stubs, and two wheel-gate tests (PyArrow round trips, + bounded batch sizes, concatenated column equality). +- `spatialrust-pipeline::StreamingPipelineIter::schema()` additive accessor. +- CI installs `pyarrow` for the Python binding job. + +## Verification (PyArrow 25 in a venv) + +- `pa.array(cloud)` → struct; field buffers are CPU-backed and + sized `N*4`, and field values match the source — zero-copy. +- `pa.RecordBatchReader.from_stream(stream)` over a 1000-point PCD yields + 256/256/256/232 batches; concatenated `x` column matches a direct read. +- `gc.collect()` after consuming shows no double free. +- pytest `-k arrow_c` passes; Rust workspace tests pass; clippy `-D warnings` + clean for the py crate. + +## Contract notes + +- The capsule destructor only acts when the capsule name is still the original + `arrow_array`/`arrow_schema`/`arrow_array_stream`; consumers that move data + mark the release callback null, so there is exactly one release per export. +- `__arrow_c_stream__` consumes the stream: the underlying iterator is moved + into the capsule, so the Python object becomes exhausted afterwards.