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

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(&self, py: Python) -> PyResult<PyRecordBatch> {

let stream = self.stream.clone();

wait_for_future(py, next_stream(stream, true))?

}

fn __next__(&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: &mut SendableRecordBatchStream,

) -> datafusion::error::Result<Option<RecordBatch>> {

stream.next().await.transpose()

stream: Arc<Mutex<SendableRecordBatchStream>>,

sync: bool,

) -> PyResult<PyRecordBatch> {

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

match poll_next_batch(&mut stream).await {

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

Ok(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"))

}

}

Err(e) => Err(PyDataFusionError::from(e))?,

}