fn to_pyarrow(&self, py: Python) -> PyResult<PyObject> {

self.batch.to_pyarrow(py)

}

fn from(batch: RecordBatch) -> Self {

Self { batch }

}

pub fn new(stream: SendableRecordBatchStream) -> Self {

Self {

stream: Arc::new(Mutex::new(stream)),

}

}

fn next(&mut self, py: Python) -> PyResult<PyRecordBatch> {

let stream = self.stream.clone();

wait_for_future(py, next_stream(stream, true))

}

fn __next__(&mut self, py: Python) -> PyResult<PyRecordBatch> {

self.next(py)

}

fn __anext__<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {

let stream = self.stream.clone();

pyo3_async_runtimes::tokio::future_into_py(py, next_stream(stream, false))

}

fn __iter__(slf: PyRef<'_, Self>) -> PyRef<'_, Self> {

slf

}

fn __aiter__(slf: PyRef<'_, Self>) -> PyRef<'_, Self> {

slf

}

stream: Arc<Mutex<SendableRecordBatchStream>>,

sync: bool,

) -> PyResult<PyRecordBatch> {

let mut stream = stream.lock().await;

match stream.next().await {

Some(Ok(batch)) => Ok(batch.into()),

Some(Err(e)) => Err(e.into()),

None => {

// Depending on whether the iteration is sync or not, we raise either a

// StopIteration or a StopAsyncIteration

if sync {

Err(PyStopIteration::new_err("stream exhausted"))

} else {

Err(PyStopAsyncIteration::new_err("stream exhausted"))

}

}

}