1use std::convert::{From, TryFrom};
63use std::ffi::CStr;
64use std::ptr::NonNull;
65use std::sync::Arc;
66
67use arrow_array::ffi;
68use arrow_array::ffi::{FFI_ArrowArray, FFI_ArrowSchema};
69use arrow_array::ffi_stream::{ArrowArrayStreamReader, FFI_ArrowArrayStream};
70use arrow_array::{
71 RecordBatch, RecordBatchIterator, RecordBatchOptions, RecordBatchReader, StructArray,
72 make_array,
73};
74use arrow_data::ArrayData;
75use arrow_schema::{ArrowError, DataType, Field, Schema, SchemaRef};
76use pyo3::exceptions::{PyTypeError, PyValueError};
77use pyo3::ffi::Py_uintptr_t;
78use pyo3::prelude::*;
79use pyo3::sync::PyOnceLock;
80use pyo3::types::{PyCapsule, PyDict, PyList, PyString, PyType};
81use pyo3::{CastError, import_exception, intern};
82
83import_exception!(pyarrow, ArrowException);
84pub type PyArrowException = ArrowException;
86
87fn to_py_err(err: ArrowError) -> PyErr {
88 PyArrowException::new_err(err.to_string())
89}
90
91pub trait FromPyArrow: Sized {
93 fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self>;
97}
98
99pub trait ToPyArrow {
101 fn to_pyarrow<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>>;
103}
104
105pub trait IntoPyArrow {
107 fn into_pyarrow<'py>(self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>>;
109}
110
111impl<T: ToPyArrow> IntoPyArrow for T {
112 fn into_pyarrow<'py>(self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
113 self.to_pyarrow(py)
114 }
115}
116
117fn validate_class(expected: &Bound<PyType>, value: &Bound<PyAny>) -> PyResult<()> {
118 if !value.is_instance(expected)? {
119 return Err(PyTypeError::new_err(format!(
120 "Expected instance of {}, got {}",
121 expected.fully_qualified_name()?,
122 value.get_type().fully_qualified_name()?
123 )));
124 }
125 Ok(())
126}
127
128impl FromPyArrow for DataType {
129 fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
130 if let Some(capsule) =
134 call_capsule_method_if_exists(value, intern!(value.py(), "__arrow_c_schema__"))?
135 {
136 let schema_ptr = extract_capsule::<FFI_ArrowSchema>(
137 &capsule,
138 c"arrow_schema",
139 "__arrow_c_schema__",
140 )?;
141 return unsafe { DataType::try_from(schema_ptr.as_ref()) }.map_err(to_py_err);
142 }
143
144 validate_class(data_type_class(value.py())?, value)?;
145
146 let mut c_schema = FFI_ArrowSchema::empty();
147 value.call_method1(
148 intern!(value.py(), "_export_to_c"),
149 (&raw mut c_schema as Py_uintptr_t,),
150 )?;
151 DataType::try_from(&c_schema).map_err(to_py_err)
152 }
153}
154
155impl ToPyArrow for DataType {
156 fn to_pyarrow<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
157 let c_schema = FFI_ArrowSchema::try_from(self).map_err(to_py_err)?;
158 data_type_class(py)?.call_method1(
159 intern!(py, "_import_from_c"),
160 (&raw const c_schema as Py_uintptr_t,),
161 )
162 }
163}
164
165impl FromPyArrow for Field {
166 fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
167 if let Some(capsule) =
171 call_capsule_method_if_exists(value, intern!(value.py(), "__arrow_c_schema__"))?
172 {
173 let schema_ptr = extract_capsule::<FFI_ArrowSchema>(
174 &capsule,
175 c"arrow_schema",
176 "__arrow_c_schema__",
177 )?;
178 return unsafe { Field::try_from(schema_ptr.as_ref()) }.map_err(to_py_err);
179 }
180
181 validate_class(field_class(value.py())?, value)?;
182
183 let mut c_schema = FFI_ArrowSchema::empty();
184 value.call_method1(
185 intern!(value.py(), "_export_to_c"),
186 (&raw mut c_schema as Py_uintptr_t,),
187 )?;
188 Field::try_from(&c_schema).map_err(to_py_err)
189 }
190}
191
192impl ToPyArrow for Field {
193 fn to_pyarrow<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
194 let c_schema = FFI_ArrowSchema::try_from(self).map_err(to_py_err)?;
195 field_class(py)?.call_method1(
196 intern!(py, "_import_from_c"),
197 (&raw const c_schema as Py_uintptr_t,),
198 )
199 }
200}
201
202impl FromPyArrow for Schema {
203 fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
204 if let Some(capsule) =
208 call_capsule_method_if_exists(value, intern!(value.py(), "__arrow_c_schema__"))?
209 {
210 let schema_ptr = extract_capsule::<FFI_ArrowSchema>(
211 &capsule,
212 c"arrow_schema",
213 "__arrow_c_schema__",
214 )?;
215 return unsafe { Schema::try_from(schema_ptr.as_ref()) }.map_err(to_py_err);
216 }
217
218 validate_class(schema_class(value.py())?, value)?;
219
220 let mut c_schema = FFI_ArrowSchema::empty();
221 value.call_method1(
222 intern!(value.py(), "_export_to_c"),
223 (&raw mut c_schema as Py_uintptr_t,),
224 )?;
225 Schema::try_from(&c_schema).map_err(to_py_err)
226 }
227}
228
229impl ToPyArrow for Schema {
230 fn to_pyarrow<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
231 let c_schema = FFI_ArrowSchema::try_from(self).map_err(to_py_err)?;
232 schema_class(py)?.call_method1(
233 intern!(py, "_import_from_c"),
234 (&raw const c_schema as Py_uintptr_t,),
235 )
236 }
237}
238
239impl FromPyArrow for ArrayData {
240 fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
241 if let Some((schema_capsule, array_capsule)) = call_arrow_c_array_method_if_exists(value)? {
245 let schema_ptr =
246 extract_capsule(&schema_capsule, c"arrow_schema", "__arrow_c_array__")?;
247 let array_ptr = extract_capsule(&array_capsule, c"arrow_array", "__arrow_c_array__")?;
248 let array = unsafe { FFI_ArrowArray::from_raw(array_ptr.as_ptr()) };
249 return unsafe { ffi::from_ffi(array, schema_ptr.as_ref()) }.map_err(to_py_err);
250 }
251
252 validate_class(array_class(value.py())?, value)?;
253
254 let mut array = FFI_ArrowArray::empty();
256 let mut schema = FFI_ArrowSchema::empty();
257
258 value.call_method1(
262 intern!(value.py(), "_export_to_c"),
263 (
264 &raw mut array as Py_uintptr_t,
265 &raw mut schema as Py_uintptr_t,
266 ),
267 )?;
268
269 unsafe { ffi::from_ffi(array, &schema) }.map_err(to_py_err)
270 }
271}
272
273impl ToPyArrow for ArrayData {
274 fn to_pyarrow<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
275 let array = FFI_ArrowArray::new(self);
276 let schema = FFI_ArrowSchema::try_from(self.data_type()).map_err(to_py_err)?;
277 array_class(py)?.call_method1(
278 intern!(py, "_import_from_c"),
279 (
280 &raw const array as Py_uintptr_t,
281 &raw const schema as Py_uintptr_t,
282 ),
283 )
284 }
285}
286
287impl<T: FromPyArrow> FromPyArrow for Vec<T> {
288 fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
289 let mut v = Vec::with_capacity(value.len().unwrap_or(0));
290 for item in value.try_iter()? {
291 v.push(T::from_pyarrow_bound(&item?)?);
292 }
293 Ok(v)
294 }
295}
296
297impl<T: ToPyArrow> ToPyArrow for Vec<T> {
298 fn to_pyarrow<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
299 self.iter()
300 .map(|v| v.to_pyarrow(py))
301 .collect::<PyResult<Vec<_>>>()?
302 .into_pyobject(py)
303 }
304}
305
306impl FromPyArrow for RecordBatch {
307 fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
308 if let Some((schema_capsule, array_capsule)) = call_arrow_c_array_method_if_exists(value)? {
313 let schema_ptr =
314 extract_capsule(&schema_capsule, c"arrow_schema", "__arrow_c_array__")?;
315 let array_ptr = extract_capsule(&array_capsule, c"arrow_array", "__arrow_c_array__")?;
316 let ffi_array = unsafe { FFI_ArrowArray::from_raw(array_ptr.as_ptr()) };
317 let array_data =
318 unsafe { ffi::from_ffi(ffi_array, schema_ptr.as_ref()) }.map_err(to_py_err)?;
319 if !matches!(array_data.data_type(), DataType::Struct(_)) {
320 return Err(PyTypeError::new_err(format!(
321 "Expected Struct type from __arrow_c_array__, found {}.",
322 array_data.data_type()
323 )));
324 }
325 let options = RecordBatchOptions::default().with_row_count(Some(array_data.len()));
326 let array = StructArray::from(array_data);
327 let schema =
330 unsafe { Arc::new(Schema::try_from(schema_ptr.as_ref()).map_err(to_py_err)?) };
331 let (_fields, columns, nulls) = array.into_parts();
332 if nulls.map(|n| n.null_count()).unwrap_or_default() != 0 {
333 return Err(PyValueError::new_err(
334 "Cannot convert nullable StructArray to RecordBatch, see StructArray documentation",
335 ));
336 }
337 return RecordBatch::try_new_with_options(schema, columns, &options).map_err(to_py_err);
338 }
339
340 validate_class(record_batch_class(value.py())?, value)?;
341 let schema = value.getattr("schema")?;
343 let schema = Arc::new(Schema::from_pyarrow_bound(&schema)?);
344
345 let arrays = value.getattr("columns")?;
346 let arrays = arrays
347 .cast::<PyList>()?
348 .iter()
349 .map(|a| Ok(make_array(ArrayData::from_pyarrow_bound(&a)?)))
350 .collect::<PyResult<_>>()?;
351
352 let row_count = value
353 .getattr("num_rows")
354 .ok()
355 .and_then(|x| x.extract().ok());
356 let options = RecordBatchOptions::default().with_row_count(row_count);
357
358 let batch =
359 RecordBatch::try_new_with_options(schema, arrays, &options).map_err(to_py_err)?;
360 Ok(batch)
361 }
362}
363
364impl ToPyArrow for RecordBatch {
365 fn to_pyarrow<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
366 let reader = RecordBatchIterator::new(vec![Ok(self.clone())], self.schema());
368 let reader: Box<dyn RecordBatchReader + Send> = Box::new(reader);
369 let py_reader = reader.into_pyarrow(py)?;
370 py_reader.call_method0(intern!(py, "read_next_batch"))
371 }
372}
373
374impl FromPyArrow for ArrowArrayStreamReader {
376 fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
377 if let Some(capsule) =
381 call_capsule_method_if_exists(value, intern!(value.py(), "__arrow_c_stream__"))?
382 {
383 let stream_ptr =
384 extract_capsule(&capsule, c"arrow_array_stream", "__arrow_c_stream__")?;
385 let stream = unsafe { FFI_ArrowArrayStream::from_raw(stream_ptr.as_ptr()) };
386
387 let stream_reader = ArrowArrayStreamReader::try_new(stream)
388 .map_err(|err| PyValueError::new_err(err.to_string()))?;
389
390 return Ok(stream_reader);
391 }
392
393 validate_class(record_batch_reader_class(value.py())?, value)?;
394
395 let mut stream = FFI_ArrowArrayStream::empty();
397
398 value.call_method1(
402 intern!(value.py(), "_export_to_c"),
403 (&raw mut stream as Py_uintptr_t,),
404 )?;
405
406 ArrowArrayStreamReader::try_new(stream)
407 .map_err(|err| PyValueError::new_err(err.to_string()))
408 }
409}
410
411impl IntoPyArrow for Box<dyn RecordBatchReader + Send> {
413 fn into_pyarrow<'py>(self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
416 let stream = FFI_ArrowArrayStream::new(self);
417 record_batch_reader_class(py)?.call_method1(
418 intern!(py, "_import_from_c"),
419 (&raw const stream as Py_uintptr_t,),
420 )
421 }
422}
423
424impl IntoPyArrow for ArrowArrayStreamReader {
426 fn into_pyarrow<'py>(self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
427 let boxed: Box<dyn RecordBatchReader + Send> = Box::new(self);
428 boxed.into_pyarrow(py)
429 }
430}
431
432#[derive(Clone)]
450pub struct Table {
451 record_batches: Vec<RecordBatch>,
452 schema: SchemaRef,
453}
454
455impl Table {
456 pub fn try_new(
457 record_batches: Vec<RecordBatch>,
458 schema: SchemaRef,
459 ) -> Result<Self, ArrowError> {
460 for record_batch in &record_batches {
461 if schema != record_batch.schema() {
462 return Err(ArrowError::SchemaError(format!(
463 "All record batches must have the same schema. \
464 Expected schema: {:?}, got schema: {:?}",
465 schema,
466 record_batch.schema()
467 )));
468 }
469 }
470 Ok(Self {
471 record_batches,
472 schema,
473 })
474 }
475
476 pub fn record_batches(&self) -> &[RecordBatch] {
477 &self.record_batches
478 }
479
480 pub fn schema(&self) -> SchemaRef {
481 self.schema.clone()
482 }
483
484 pub fn into_inner(self) -> (Vec<RecordBatch>, SchemaRef) {
485 (self.record_batches, self.schema)
486 }
487}
488
489impl TryFrom<Box<dyn RecordBatchReader>> for Table {
490 type Error = ArrowError;
491
492 fn try_from(value: Box<dyn RecordBatchReader>) -> Result<Self, ArrowError> {
493 let schema = value.schema();
494 let batches = value.collect::<Result<Vec<_>, _>>()?;
495 Self::try_new(batches, schema)
496 }
497}
498
499impl FromPyArrow for Table {
501 fn from_pyarrow_bound(ob: &Bound<PyAny>) -> PyResult<Self> {
502 let reader: Box<dyn RecordBatchReader> =
503 Box::new(ArrowArrayStreamReader::from_pyarrow_bound(ob)?);
504 Self::try_from(reader).map_err(|err| PyValueError::new_err(err.to_string()))
505 }
506}
507
508impl IntoPyArrow for Table {
510 fn into_pyarrow(self, py: Python) -> PyResult<Bound<PyAny>> {
511 let py_batches = PyList::new(py, self.record_batches.into_iter().map(PyArrowType))?;
512 let py_schema = PyArrowType(Arc::unwrap_or_clone(self.schema));
513
514 let kwargs = PyDict::new(py);
515 kwargs.set_item("schema", py_schema)?;
516
517 table_class(py)?.call_method("from_batches", (py_batches,), Some(&kwargs))
518 }
519}
520
521fn array_class(py: Python<'_>) -> PyResult<&Bound<'_, PyType>> {
522 static TYPE: PyOnceLock<Py<PyType>> = PyOnceLock::new();
523 TYPE.import(py, "pyarrow", "Array")
524}
525
526fn record_batch_class(py: Python<'_>) -> PyResult<&Bound<'_, PyType>> {
527 static TYPE: PyOnceLock<Py<PyType>> = PyOnceLock::new();
528 TYPE.import(py, "pyarrow", "RecordBatch")
529}
530
531fn record_batch_reader_class(py: Python<'_>) -> PyResult<&Bound<'_, PyType>> {
532 static TYPE: PyOnceLock<Py<PyType>> = PyOnceLock::new();
533 TYPE.import(py, "pyarrow", "RecordBatchReader")
534}
535
536fn data_type_class(py: Python<'_>) -> PyResult<&Bound<'_, PyType>> {
537 static TYPE: PyOnceLock<Py<PyType>> = PyOnceLock::new();
538 TYPE.import(py, "pyarrow", "DataType")
539}
540
541fn field_class(py: Python<'_>) -> PyResult<&Bound<'_, PyType>> {
542 static TYPE: PyOnceLock<Py<PyType>> = PyOnceLock::new();
543 TYPE.import(py, "pyarrow", "Field")
544}
545
546fn schema_class(py: Python<'_>) -> PyResult<&Bound<'_, PyType>> {
547 static TYPE: PyOnceLock<Py<PyType>> = PyOnceLock::new();
548 TYPE.import(py, "pyarrow", "Schema")
549}
550
551fn table_class(py: Python<'_>) -> PyResult<&Bound<'_, PyType>> {
552 static TYPE: PyOnceLock<Py<PyType>> = PyOnceLock::new();
553 TYPE.import(py, "pyarrow", "Table")
554}
555
556#[derive(Debug)]
562pub struct PyArrowType<T>(pub T);
563
564impl<T: FromPyArrow> FromPyObject<'_, '_> for PyArrowType<T> {
565 type Error = PyErr;
566
567 fn extract(value: Borrowed<'_, '_, PyAny>) -> PyResult<Self> {
568 Ok(Self(T::from_pyarrow_bound(&value)?))
569 }
570}
571
572impl<'py, T: IntoPyArrow> IntoPyObject<'py> for PyArrowType<T> {
573 type Target = PyAny;
574
575 type Output = Bound<'py, Self::Target>;
576
577 type Error = PyErr;
578
579 fn into_pyobject(self, py: Python<'py>) -> PyResult<Self::Output> {
580 self.0.into_pyarrow(py)
581 }
582}
583
584impl<T> From<T> for PyArrowType<T> {
585 fn from(s: T) -> Self {
586 Self(s)
587 }
588}
589
590fn call_capsule_method_if_exists<'py>(
591 object: &Bound<'py, PyAny>,
592 method_name: &Bound<'py, PyString>,
593) -> PyResult<Option<Bound<'py, PyCapsule>>> {
594 let Some(method) = object.getattr_opt(method_name)? else {
595 return Ok(None);
596 };
597 Ok(Some(method.call0()?.extract().map_err(
598 |e: CastError| {
599 wrapping_type_error(
600 object.py(),
601 e.into(),
602 format!("Expected {method_name} to return a capsule."),
603 )
604 },
605 )?))
606}
607
608fn call_arrow_c_array_method_if_exists<'py>(
609 object: &Bound<'py, PyAny>,
610) -> PyResult<Option<(Bound<'py, PyCapsule>, Bound<'py, PyCapsule>)>> {
611 let Some(method) = object.getattr_opt(intern!(object.py(), "__arrow_c_array__"))? else {
612 return Ok(None);
613 };
614 Ok(Some(method.call0()?.extract().map_err(|e| {
615 wrapping_type_error(
616 object.py(),
617 e,
618 "Expected __arrow_c_array__ to return a tuple of (schema, array) capsules.".into(),
619 )
620 })?))
621}
622
623fn extract_capsule<T>(
624 capsule: &Bound<PyCapsule>,
625 capsule_name: &CStr,
626 method_name: &'static str,
627) -> PyResult<NonNull<T>> {
628 Ok(capsule
629 .pointer_checked(Some(capsule_name))
630 .map_err(|e| {
631 wrapping_type_error(
632 capsule.py(),
633 e,
634 format!(
635 "Expected {method_name} to return a {} capsule.",
636 capsule_name.to_str().unwrap(),
637 ),
638 )
639 })?
640 .cast::<T>())
641}
642
643fn wrapping_type_error(py: Python<'_>, error: PyErr, message: String) -> PyErr {
644 let e = PyTypeError::new_err(message);
645 e.set_cause(py, Some(error));
646 e
647}