Function _test_round_trip_export

arrow-array/src/ffi_stream.rs:428–471

(batch: RecordBatch, schema: Arc<Schema>)

    }

    fn _test_round_trip_export(batch: RecordBatch, schema: Arc<Schema>) -> Result<()> {
        let iter = Box::new(vec![batch.clone(), batch.clone()].into_iter().map(Ok)) as _;

        let reader = TestRecordBatchReader::new(schema.clone(), iter);

        // Export a `RecordBatchReader` through `FFI_ArrowArrayStream`
        let mut ffi_stream = FFI_ArrowArrayStream::new(reader);

        // Get schema from `FFI_ArrowArrayStream`
        let mut ffi_schema = FFI_ArrowSchema::empty();
        let ret_code = unsafe { get_schema(&mut ffi_stream, &mut ffi_schema) };
        assert_eq!(ret_code, 0);

        let exported_schema = Schema::try_from(&ffi_schema).unwrap();
        assert_eq!(&exported_schema, schema.as_ref());

        // Get array from `FFI_ArrowArrayStream`
        let mut produced_batches = vec![];
        loop {
            let mut ffi_array = FFI_ArrowArray::empty();
            let ret_code = unsafe { get_next(&mut ffi_stream, &mut ffi_array) };
            assert_eq!(ret_code, 0);

            // The end of stream has been reached
            if ffi_array.is_released() {
                break;
            }

            let array = unsafe { from_ffi(ffi_array, &ffi_schema) }.unwrap();
            let len = array.len();

            let record_batch = RecordBatch::try_new_with_options(
                SchemaRef::from(exported_schema.clone()),
                StructArray::from(array).into_parts().1,
                &RecordBatchOptions::new().with_row_count(Some(len)),
            )
            .unwrap();
            produced_batches.push(record_batch);
        }

        assert_eq!(produced_batches, vec![batch.clone(), batch]);

        Ok(())
    }

    fn _test_round_trip_import(batch: RecordBatch, schema: Arc<Schema>) -> Result<()> {
        let iter = Box::new(vec![batch.clone(), batch.clone()].into_iter().map(Ok)) as _;