use std::convert::{From, TryFrom};
use std::ffi::CStr;
use std::ptr::NonNull;
use std::sync::Arc;
use arrow_array::ffi;
use arrow_array::ffi::{FFI_ArrowArray, FFI_ArrowSchema};
use arrow_array::ffi_stream::{FFI_ArrowArrayStream, RecordBatchStreamReader};
use arrow_array::{
RecordBatch, RecordBatchIterator, RecordBatchOptions, RecordBatchReader, StructArray,
make_array,
};
use arrow_data::ArrayData;
use arrow_schema::{ArrowError, DataType, Field, Schema, SchemaRef};
use pyo3::exceptions::{PyTypeError, PyValueError};
use pyo3::ffi::Py_uintptr_t;
use pyo3::prelude::*;
use pyo3::sync::PyOnceLock;
use pyo3::types::{PyCapsule, PyDict, PyList, PyString, PyType};
use pyo3::{CastError, import_exception, intern};
#[cfg(feature = "experimental-inspect")]
use pyo3::{
inspect::PyStaticExpr, type_hint_identifier, type_hint_subscript, type_hint_union,
type_object::PyTypeInfo,
};
macro_rules! type_hint {
($name:ident = $hint:expr) => {
#[cfg(feature = "experimental-inspect")]
const $name: PyStaticExpr = $hint;
};
}
import_exception!(pyarrow, ArrowException);
pub type PyArrowException = ArrowException;
fn to_py_err(err: ArrowError) -> PyErr {
PyArrowException::new_err(err.to_string())
}
#[cfg(feature = "experimental-inspect")]
const ARRAY_STREAM_INPUT_TYPE: PyStaticExpr = type_hint_union!(
type_hint_identifier!("pyarrow", "RecordBatchReader"),
type_hint_identifier!("pyarrow", "Table")
);
pub trait FromPyArrowUnchecked: Sized {
unsafe fn from_pyarrow_bound_unchecked(value: &Bound<PyAny>) -> PyResult<Self>;
}
pub trait FromPyArrow: Sized {
#[cfg(feature = "experimental-inspect")]
const INPUT_TYPE: PyStaticExpr = type_hint_identifier!("_typeshed", "Incomplete");
fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self>;
}
pub trait ToPyArrow {
#[cfg(feature = "experimental-inspect")]
const OUTPUT_TYPE: PyStaticExpr = type_hint_identifier!("_typeshed", "Incomplete");
fn to_pyarrow<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>>;
}
pub trait IntoPyArrow {
#[cfg(feature = "experimental-inspect")]
const OUTPUT_TYPE: PyStaticExpr = type_hint_identifier!("_typeshed", "Incomplete");
fn into_pyarrow(self, py: Python<'_>) -> PyResult<Bound<'_, PyAny>>;
}
impl<T: ToPyArrow> IntoPyArrow for T {
type_hint!(OUTPUT_TYPE = <T as ToPyArrow>::OUTPUT_TYPE);
fn into_pyarrow(self, py: Python<'_>) -> PyResult<Bound<'_, PyAny>> {
self.to_pyarrow(py)
}
}
fn validate_class(expected: &Bound<PyType>, value: &Bound<PyAny>) -> PyResult<()> {
if !value.is_instance(expected)? {
return Err(PyTypeError::new_err(format!(
"Expected instance of {}, got {}",
expected.fully_qualified_name()?,
value.get_type().fully_qualified_name()?
)));
}
Ok(())
}
impl FromPyArrow for DataType {
type_hint!(INPUT_TYPE = type_hint_identifier!("pyarrow", "DataType"));
fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
if let Some(capsule) =
call_capsule_method_if_exists(value, intern!(value.py(), "__arrow_c_schema__"))?
{
let schema_ptr = extract_capsule::<FFI_ArrowSchema>(
&capsule,
c"arrow_schema",
"__arrow_c_schema__",
)?;
return unsafe { DataType::try_from(schema_ptr.as_ref()) }.map_err(to_py_err);
}
validate_class(data_type_class(value.py())?, value)?;
let mut c_schema = FFI_ArrowSchema::empty();
value.call_method1(
intern!(value.py(), "_export_to_c"),
(&raw mut c_schema as Py_uintptr_t,),
)?;
DataType::try_from(&c_schema).map_err(to_py_err)
}
}
impl ToPyArrow for DataType {
type_hint!(OUTPUT_TYPE = type_hint_identifier!("pyarrow", "DataType"));
fn to_pyarrow<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
let c_schema = FFI_ArrowSchema::try_from(self).map_err(to_py_err)?;
data_type_class(py)?.call_method1(
intern!(py, "_import_from_c"),
(&raw const c_schema as Py_uintptr_t,),
)
}
}
impl FromPyArrow for Field {
type_hint!(INPUT_TYPE = type_hint_identifier!("pyarrow", "Field"));
fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
if let Some(capsule) =
call_capsule_method_if_exists(value, intern!(value.py(), "__arrow_c_schema__"))?
{
let schema_ptr = extract_capsule::<FFI_ArrowSchema>(
&capsule,
c"arrow_schema",
"__arrow_c_schema__",
)?;
return unsafe { Field::try_from(schema_ptr.as_ref()) }.map_err(to_py_err);
}
validate_class(field_class(value.py())?, value)?;
let mut c_schema = FFI_ArrowSchema::empty();
value.call_method1(
intern!(value.py(), "_export_to_c"),
(&raw mut c_schema as Py_uintptr_t,),
)?;
Field::try_from(&c_schema).map_err(to_py_err)
}
}
impl ToPyArrow for Field {
type_hint!(OUTPUT_TYPE = type_hint_identifier!("pyarrow", "Field"));
fn to_pyarrow<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
let c_schema = FFI_ArrowSchema::try_from(self).map_err(to_py_err)?;
field_class(py)?.call_method1(
intern!(py, "_import_from_c"),
(&raw const c_schema as Py_uintptr_t,),
)
}
}
impl FromPyArrow for Schema {
type_hint!(INPUT_TYPE = type_hint_identifier!("pyarrow", "Schema"));
fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
if let Some(capsule) =
call_capsule_method_if_exists(value, intern!(value.py(), "__arrow_c_schema__"))?
{
let schema_ptr = extract_capsule::<FFI_ArrowSchema>(
&capsule,
c"arrow_schema",
"__arrow_c_schema__",
)?;
return unsafe { Schema::try_from(schema_ptr.as_ref()) }.map_err(to_py_err);
}
validate_class(schema_class(value.py())?, value)?;
let mut c_schema = FFI_ArrowSchema::empty();
value.call_method1(
intern!(value.py(), "_export_to_c"),
(&raw mut c_schema as Py_uintptr_t,),
)?;
Schema::try_from(&c_schema).map_err(to_py_err)
}
}
impl ToPyArrow for Schema {
type_hint!(OUTPUT_TYPE = type_hint_identifier!("pyarrow", "Schema"));
fn to_pyarrow<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
let c_schema = FFI_ArrowSchema::try_from(self).map_err(to_py_err)?;
schema_class(py)?.call_method1(
intern!(py, "_import_from_c"),
(&raw const c_schema as Py_uintptr_t,),
)
}
}
impl FromPyArrowUnchecked for ArrayData {
unsafe fn from_pyarrow_bound_unchecked(value: &Bound<PyAny>) -> PyResult<Self> {
if let Some((schema_capsule, array_capsule)) = call_arrow_c_array_method_if_exists(value)? {
let schema_ptr =
extract_capsule(&schema_capsule, c"arrow_schema", "__arrow_c_array__")?;
let array_ptr = extract_capsule(&array_capsule, c"arrow_array", "__arrow_c_array__")?;
let array = unsafe { FFI_ArrowArray::from_raw(array_ptr.as_ptr()) };
return unsafe { ffi::from_ffi(array, schema_ptr.as_ref()) }.map_err(to_py_err);
}
validate_class(array_class(value.py())?, value)?;
let mut array = FFI_ArrowArray::empty();
let mut schema = FFI_ArrowSchema::empty();
value.call_method1(
intern!(value.py(), "_export_to_c"),
(
&raw mut array as Py_uintptr_t,
&raw mut schema as Py_uintptr_t,
),
)?;
unsafe { ffi::from_ffi(array, &schema) }.map_err(to_py_err)
}
}
impl FromPyArrow for ArrayData {
type_hint!(INPUT_TYPE = type_hint_identifier!("pyarrow", "Array"));
fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
let data = unsafe { ArrayData::from_pyarrow_bound_unchecked(value)? };
data.validate_full().map_err(to_py_err)?;
Ok(data)
}
}
impl ToPyArrow for ArrayData {
type_hint!(OUTPUT_TYPE = type_hint_identifier!("pyarrow", "Array"));
fn to_pyarrow<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
let array = FFI_ArrowArray::new(self);
let schema = FFI_ArrowSchema::try_from(self.data_type()).map_err(to_py_err)?;
array_class(py)?.call_method1(
intern!(py, "_import_from_c"),
(
&raw const array as Py_uintptr_t,
&raw const schema as Py_uintptr_t,
),
)
}
}
impl<T: FromPyArrow> FromPyArrow for Vec<T> {
type_hint!(
INPUT_TYPE = type_hint_subscript!(
type_hint_identifier!("collections.abc", "Iterable"),
<T as FromPyArrow>::INPUT_TYPE
)
);
fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
let mut v = Vec::with_capacity(value.len().unwrap_or(0));
for item in value.try_iter()? {
v.push(T::from_pyarrow_bound(&item?)?);
}
Ok(v)
}
}
impl<T: ToPyArrow> ToPyArrow for Vec<T> {
type_hint!(
OUTPUT_TYPE = type_hint_subscript!(PyList::TYPE_HINT, <T as ToPyArrow>::OUTPUT_TYPE)
);
fn to_pyarrow<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
self.iter()
.map(|v| v.to_pyarrow(py))
.collect::<PyResult<Vec<_>>>()?
.into_pyobject(py)
}
}
fn record_batch_from_pyarrow_bound_impl(
value: &Bound<PyAny>,
validate: bool,
) -> PyResult<RecordBatch> {
if let Some((schema_capsule, array_capsule)) = call_arrow_c_array_method_if_exists(value)? {
let schema_ptr = extract_capsule(&schema_capsule, c"arrow_schema", "__arrow_c_array__")?;
let array_ptr = extract_capsule(&array_capsule, c"arrow_array", "__arrow_c_array__")?;
let ffi_array = unsafe { FFI_ArrowArray::from_raw(array_ptr.as_ptr()) };
let array_data =
unsafe { ffi::from_ffi(ffi_array, schema_ptr.as_ref()) }.map_err(to_py_err)?;
if validate {
array_data.validate_full().map_err(to_py_err)?;
}
if !matches!(array_data.data_type(), DataType::Struct(_)) {
return Err(PyTypeError::new_err(format!(
"Expected Struct type from __arrow_c_array__, found {}.",
array_data.data_type()
)));
}
let options = RecordBatchOptions::default().with_row_count(Some(array_data.len()));
let array = StructArray::from(array_data);
let schema = unsafe { Arc::new(Schema::try_from(schema_ptr.as_ref()).map_err(to_py_err)?) };
let (_fields, columns, nulls) = array.into_parts();
if nulls.map(|n| n.null_count()).unwrap_or_default() != 0 {
return Err(PyValueError::new_err(
"Cannot convert nullable StructArray to RecordBatch, see StructArray documentation",
));
}
return RecordBatch::try_new_with_options(schema, columns, &options).map_err(to_py_err);
}
validate_class(record_batch_class(value.py())?, value)?;
let schema = value.getattr("schema")?;
let schema = Arc::new(Schema::from_pyarrow_bound(&schema)?);
let arrays = value.getattr("columns")?;
let arrays = arrays
.cast::<PyList>()?
.iter()
.map(|a| {
let data = unsafe { ArrayData::from_pyarrow_bound_unchecked(&a)? };
if validate {
data.validate_full().map_err(to_py_err)?;
}
Ok(make_array(data))
})
.collect::<PyResult<_>>()?;
let row_count = value
.getattr("num_rows")
.ok()
.and_then(|x| x.extract().ok());
let options = RecordBatchOptions::default().with_row_count(row_count);
RecordBatch::try_new_with_options(schema, arrays, &options).map_err(to_py_err)
}
impl FromPyArrow for RecordBatch {
type_hint!(INPUT_TYPE = type_hint_identifier!("pyarrow", "RecordBatch"));
fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
record_batch_from_pyarrow_bound_impl(value, true)
}
}
impl FromPyArrowUnchecked for RecordBatch {
unsafe fn from_pyarrow_bound_unchecked(value: &Bound<PyAny>) -> PyResult<Self> {
record_batch_from_pyarrow_bound_impl(value, false)
}
}
impl ToPyArrow for RecordBatch {
type_hint!(OUTPUT_TYPE = type_hint_identifier!("pyarrow", "RecordBatch"));
fn to_pyarrow<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
let reader = RecordBatchIterator::new(vec![Ok(self.clone())], self.schema());
let reader: Box<dyn RecordBatchReader + Send> = Box::new(reader);
let py_reader = reader.into_pyarrow(py)?;
py_reader.call_method0(intern!(py, "read_next_batch"))
}
}
impl FromPyArrow for RecordBatchStreamReader {
type_hint!(INPUT_TYPE = ARRAY_STREAM_INPUT_TYPE);
fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
if let Some(capsule) =
call_capsule_method_if_exists(value, intern!(value.py(), "__arrow_c_stream__"))?
{
let stream_ptr =
extract_capsule(&capsule, c"arrow_array_stream", "__arrow_c_stream__")?;
let stream = unsafe { FFI_ArrowArrayStream::from_raw(stream_ptr.as_ptr()) };
let stream_reader = RecordBatchStreamReader::try_new(stream)
.map_err(|err| PyValueError::new_err(err.to_string()))?;
return Ok(stream_reader);
}
validate_class(record_batch_reader_class(value.py())?, value)?;
let mut stream = FFI_ArrowArrayStream::empty();
value.call_method1(
intern!(value.py(), "_export_to_c"),
(&raw mut stream as Py_uintptr_t,),
)?;
RecordBatchStreamReader::try_new(stream)
.map_err(|err| PyValueError::new_err(err.to_string()))
}
}
impl IntoPyArrow for Box<dyn RecordBatchReader + Send> {
type_hint!(OUTPUT_TYPE = type_hint_identifier!("pyarrow", "RecordBatchReader"));
fn into_pyarrow(self, py: Python<'_>) -> PyResult<Bound<'_, PyAny>> {
let stream = FFI_ArrowArrayStream::new(self);
record_batch_reader_class(py)?.call_method1(
intern!(py, "_import_from_c"),
(&raw const stream as Py_uintptr_t,),
)
}
}
impl IntoPyArrow for RecordBatchStreamReader {
type_hint!(OUTPUT_TYPE = type_hint_identifier!("pyarrow", "RecordBatchReader"));
fn into_pyarrow(self, py: Python<'_>) -> PyResult<Bound<'_, PyAny>> {
let boxed: Box<dyn RecordBatchReader + Send> = Box::new(self);
boxed.into_pyarrow(py)
}
}
#[derive(Clone)]
pub struct Table {
record_batches: Vec<RecordBatch>,
schema: SchemaRef,
}
impl Table {
pub fn try_new(
record_batches: Vec<RecordBatch>,
schema: SchemaRef,
) -> Result<Self, ArrowError> {
for record_batch in &record_batches {
if schema != record_batch.schema() {
return Err(ArrowError::SchemaError(format!(
"All record batches must have the same schema. \
Expected schema: {:?}, got schema: {:?}",
schema,
record_batch.schema()
)));
}
}
Ok(Self {
record_batches,
schema,
})
}
pub fn record_batches(&self) -> &[RecordBatch] {
&self.record_batches
}
pub fn schema(&self) -> SchemaRef {
self.schema.clone()
}
pub fn into_inner(self) -> (Vec<RecordBatch>, SchemaRef) {
(self.record_batches, self.schema)
}
}
impl TryFrom<Box<dyn RecordBatchReader>> for Table {
type Error = ArrowError;
fn try_from(value: Box<dyn RecordBatchReader>) -> Result<Self, ArrowError> {
let schema = value.schema();
let batches = value.collect::<Result<Vec<_>, _>>()?;
Self::try_new(batches, schema)
}
}
impl FromPyArrow for Table {
type_hint!(INPUT_TYPE = ARRAY_STREAM_INPUT_TYPE);
fn from_pyarrow_bound(ob: &Bound<PyAny>) -> PyResult<Self> {
let reader: Box<dyn RecordBatchReader> =
Box::new(RecordBatchStreamReader::from_pyarrow_bound(ob)?);
Self::try_from(reader).map_err(|err| PyValueError::new_err(err.to_string()))
}
}
impl IntoPyArrow for Table {
type_hint!(OUTPUT_TYPE = type_hint_identifier!("pyarrow", "Table"));
fn into_pyarrow(self, py: Python) -> PyResult<Bound<PyAny>> {
let py_batches = PyList::new(py, self.record_batches.into_iter().map(PyArrowType))?;
let py_schema = PyArrowType(Arc::unwrap_or_clone(self.schema));
let kwargs = PyDict::new(py);
kwargs.set_item("schema", py_schema)?;
table_class(py)?.call_method("from_batches", (py_batches,), Some(&kwargs))
}
}
fn array_class(py: Python<'_>) -> PyResult<&Bound<'_, PyType>> {
static TYPE: PyOnceLock<Py<PyType>> = PyOnceLock::new();
TYPE.import(py, "pyarrow", "Array")
}
fn record_batch_class(py: Python<'_>) -> PyResult<&Bound<'_, PyType>> {
static TYPE: PyOnceLock<Py<PyType>> = PyOnceLock::new();
TYPE.import(py, "pyarrow", "RecordBatch")
}
fn record_batch_reader_class(py: Python<'_>) -> PyResult<&Bound<'_, PyType>> {
static TYPE: PyOnceLock<Py<PyType>> = PyOnceLock::new();
TYPE.import(py, "pyarrow", "RecordBatchReader")
}
fn data_type_class(py: Python<'_>) -> PyResult<&Bound<'_, PyType>> {
static TYPE: PyOnceLock<Py<PyType>> = PyOnceLock::new();
TYPE.import(py, "pyarrow", "DataType")
}
fn field_class(py: Python<'_>) -> PyResult<&Bound<'_, PyType>> {
static TYPE: PyOnceLock<Py<PyType>> = PyOnceLock::new();
TYPE.import(py, "pyarrow", "Field")
}
fn schema_class(py: Python<'_>) -> PyResult<&Bound<'_, PyType>> {
static TYPE: PyOnceLock<Py<PyType>> = PyOnceLock::new();
TYPE.import(py, "pyarrow", "Schema")
}
fn table_class(py: Python<'_>) -> PyResult<&Bound<'_, PyType>> {
static TYPE: PyOnceLock<Py<PyType>> = PyOnceLock::new();
TYPE.import(py, "pyarrow", "Table")
}
#[derive(Debug)]
pub struct PyArrowType<T>(pub T);
impl<T: FromPyArrow> FromPyObject<'_, '_> for PyArrowType<T> {
type Error = PyErr;
type_hint!(INPUT_TYPE = <T as FromPyArrow>::INPUT_TYPE);
fn extract(value: Borrowed<'_, '_, PyAny>) -> PyResult<Self> {
Ok(Self(T::from_pyarrow_bound(&value)?))
}
}
impl<'py, T: IntoPyArrow> IntoPyObject<'py> for PyArrowType<T> {
type Target = PyAny;
type Output = Bound<'py, Self::Target>;
type Error = PyErr;
type_hint!(OUTPUT_TYPE = <T as IntoPyArrow>::OUTPUT_TYPE);
fn into_pyobject(self, py: Python<'py>) -> PyResult<Self::Output> {
self.0.into_pyarrow(py)
}
}
impl<T> From<T> for PyArrowType<T> {
fn from(s: T) -> Self {
Self(s)
}
}
fn call_capsule_method_if_exists<'py>(
object: &Bound<'py, PyAny>,
method_name: &Bound<'py, PyString>,
) -> PyResult<Option<Bound<'py, PyCapsule>>> {
let Some(method) = object.getattr_opt(method_name)? else {
return Ok(None);
};
Ok(Some(method.call0()?.extract().map_err(
|e: CastError| {
wrapping_type_error(
object.py(),
e.into(),
format!("Expected {method_name} to return a capsule."),
)
},
)?))
}
fn call_arrow_c_array_method_if_exists<'py>(
object: &Bound<'py, PyAny>,
) -> PyResult<Option<(Bound<'py, PyCapsule>, Bound<'py, PyCapsule>)>> {
let Some(method) = object.getattr_opt(intern!(object.py(), "__arrow_c_array__"))? else {
return Ok(None);
};
Ok(Some(method.call0()?.extract().map_err(|e| {
wrapping_type_error(
object.py(),
e,
"Expected __arrow_c_array__ to return a tuple of (schema, array) capsules.".into(),
)
})?))
}
fn extract_capsule<T>(
capsule: &Bound<PyCapsule>,
capsule_name: &CStr,
method_name: &'static str,
) -> PyResult<NonNull<T>> {
Ok(capsule
.pointer_checked(Some(capsule_name))
.map_err(|e| {
wrapping_type_error(
capsule.py(),
e,
format!(
"Expected {method_name} to return a {} capsule.",
capsule_name.to_str().unwrap(),
),
)
})?
.cast::<T>())
}
fn wrapping_type_error(py: Python<'_>, error: PyErr, message: String) -> PyErr {
let e = PyTypeError::new_err(message);
e.set_cause(py, Some(error));
e
}
#[cfg(all(test, feature = "experimental-inspect"))]
mod introspection_tests {
use super::*;
use pyo3::{FromPyObject, IntoPyObject};
fn input_type<T: FromPyArrow>() -> String {
<PyArrowType<T> as FromPyObject<'_, '_>>::INPUT_TYPE.to_string()
}
fn output_type<T: IntoPyArrow>() -> String
where
PyArrowType<T>: for<'py> IntoPyObject<'py>,
{
<PyArrowType<T> as IntoPyObject<'_>>::OUTPUT_TYPE.to_string()
}
#[test]
fn scalar_types_map_to_their_pyarrow_class() {
assert_eq!(input_type::<DataType>(), "pyarrow.DataType");
assert_eq!(output_type::<DataType>(), "pyarrow.DataType");
assert_eq!(input_type::<Field>(), "pyarrow.Field");
assert_eq!(output_type::<Field>(), "pyarrow.Field");
assert_eq!(input_type::<Schema>(), "pyarrow.Schema");
assert_eq!(output_type::<Schema>(), "pyarrow.Schema");
assert_eq!(input_type::<RecordBatch>(), "pyarrow.RecordBatch");
assert_eq!(output_type::<RecordBatch>(), "pyarrow.RecordBatch");
}
#[test]
fn array_data_maps_to_pyarrow_array() {
assert_eq!(input_type::<ArrayData>(), "pyarrow.Array");
assert_eq!(output_type::<ArrayData>(), "pyarrow.Array");
}
#[test]
fn vec_is_iterable_in_and_list_out() {
assert_eq!(
input_type::<Vec<RecordBatch>>(),
"collections.abc.Iterable[pyarrow.RecordBatch]"
);
assert_eq!(
output_type::<Vec<RecordBatch>>(),
"builtins.list[pyarrow.RecordBatch]"
);
}
#[test]
fn readers_and_tables_map_to_their_pyarrow_class() {
assert_eq!(
input_type::<RecordBatchStreamReader>(),
"pyarrow.RecordBatchReader | pyarrow.Table"
);
assert_eq!(
output_type::<RecordBatchStreamReader>(),
"pyarrow.RecordBatchReader"
);
assert_eq!(
output_type::<Box<dyn RecordBatchReader + Send>>(),
"pyarrow.RecordBatchReader"
);
assert_eq!(
input_type::<Table>(),
"pyarrow.RecordBatchReader | pyarrow.Table"
);
assert_eq!(output_type::<Table>(), "pyarrow.Table");
}
}