Repository: jorgecarleitao/arrow2
Pull request #1100 (2022-06-24 15:25:52)

Reduced need of `unsafe` in FFI
This PR simplifies the API of FFI:

* allocating the FFI structs on the heap is no longer required
* exporting is now `safe` since we only need the struct to be on the stack
* use `Box` on `Bytes`, further simplifying this core struct.
* use `write` instead of `write_unaligned`
* cleaup the integration test code

# Backward incompatible changes

We no longer require a pointer to export - just export the struct to the stack and Box it if the other end of the interface requires a pointer. If the other end is the one allocating an empty struct, then write the exported struct to the pointer (which is `unsafe` as the receiving end must produce a valid pointer ^^).


codecov[bot]: # [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1100?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) Report
> Merging [#1100](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1100?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (25ab889) into [main](https://codecov.io/gh/jorgecarleitao/arrow2/commit/acd14f87bf10001493761d9d4c8da33949a813b3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (acd14f8) will **increase** coverage by `0.03%`.
> The diff coverage is `98.30%`.

```diff
@@            Coverage Diff             @@
##             main    #1100      +/-   ##
==========================================
+ Coverage   81.01%   81.04%   +0.03%     
==========================================
  Files         367      366       -1     
  Lines       35340    35320      -20     
==========================================
- Hits        28629    28626       -3     
+ Misses       6711     6694      -17     
```


| [Impacted Files](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1100?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | Coverage Δ | |
|---|---|---|
| [src/buffer/bytes.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1100/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2J1ZmZlci9ieXRlcy5ycw==) | `98.11% <95.65%> (+23.97%)` | :arrow_up: |
| [src/ffi/array.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1100/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2ZmaS9hcnJheS5ycw==) | `89.67% <100.00%> (+0.43%)` | :arrow_up: |
| [src/ffi/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1100/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2ZmaS9tb2QucnM=) | `100.00% <100.00%> (ø)` | |
| [src/ffi/stream.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1100/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2ZmaS9zdHJlYW0ucnM=) | `67.80% <100.00%> (ø)` | |
| [src/bitmap/utils/slice\_iterator.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1100/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2JpdG1hcC91dGlscy9zbGljZV9pdGVyYXRvci5ycw==) | `97.56% <0.00%> (-1.22%)` | :arrow_down: |
| [src/array/binary/mod.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1100/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2JpbmFyeS9tb2QucnM=) | `90.04% <0.00%> (+1.24%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1100?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1100?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Last update [acd14f8...25ab889](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1100?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao).


Patch:
diff --git a/.github/workflows/integration-parquet.yml b/.github/workflows/integration-parquet.yml
index 9e253a7491..53a76d5b52 100644
--- a/.github/workflows/integration-parquet.yml
+++ b/.github/workflows/integration-parquet.yml
@@ -46,7 +46,7 @@ jobs:
           python -m venv venv
           source venv/bin/activate
           pip install --upgrade pip
-          pip install pyarrow==6 pyspark==3
+          pip install pyarrow==8 pyspark==3
           python main.py
           # test against spark
           python main_spark.py
diff --git a/examples/ffi.rs b/examples/ffi.rs
index 84a65599e4..ed5b9cd87e 100644
--- a/examples/ffi.rs
+++ b/examples/ffi.rs
@@ -3,18 +3,19 @@ use arrow2::datatypes::Field;
 use arrow2::error::Result;
 use arrow2::ffi;
 
-unsafe fn export(
-    array: Box<dyn Array>,
-    array_ptr: *mut ffi::ArrowArray,
-    schema_ptr: *mut ffi::ArrowSchema,
-) {
-    // exporting an array requires an associated field so that the consumer knows its datatype
+fn export(array: Box<dyn Array>) -> (ffi::ArrowArray, ffi::ArrowSchema) {
+    // importing an array requires an associated field so that the consumer knows its datatype.
+    // Thus, we need to export both
     let field = Field::new("a", array.data_type().clone(), true);
-    ffi::export_array_to_c(array, array_ptr);
-    ffi::export_field_to_c(&field, schema_ptr);
+    (
+        ffi::export_array_to_c(array),
+        ffi::export_field_to_c(&field),
+    )
 }
 
-unsafe fn import(array: Box<ffi::ArrowArray>, schema: &ffi::ArrowSchema) -> Result<Box<dyn Array>> {
+/// # Safety
+/// `ArrowArray` and `ArrowSchema` must be valid
+unsafe fn import(array: ffi::ArrowArray, schema: &ffi::ArrowSchema) -> Result<Box<dyn Array>> {
     let field = ffi::import_field_from_c(schema)?;
     ffi::import_array_from_c(array, field.data_type)
 }
@@ -23,19 +24,14 @@ fn main() -> Result<()> {
     // let's assume that we have an array:
     let array = PrimitiveArray::<i32>::from([Some(1), None, Some(123)]).boxed();
 
-    // the goal is to export this array and import it back via FFI.
-    // to import, we initialize the structs that will receive the data
-    let mut array_ptr = Box::new(ffi::ArrowArray::empty());
-    let mut schema_ptr = Box::new(ffi::ArrowSchema::empty());
+    // here we export - `array_ffi` and `schema_ffi` are the structs of the C data interface
+    let (array_ffi, schema_ffi) = export(array.clone());
 
-    // this is where a producer (in this case also us ^_^) writes to the pointers' location.
-    // `array` here could be anything or not even be available, if this was e.g. from Python.
-    // Safety: we just allocated the pointers
-    unsafe { export(array.clone(), &mut *array_ptr, &mut *schema_ptr) };
+    // here we import them. Often the structs are wrapped in a pointer. In that case you
+    // need to read the pointer to the stack.
 
-    // and finally interpret the written memory into a new array.
     // Safety: we used `export`, which is a valid exporter to the C data interface
-    let new_array = unsafe { import(array_ptr, schema_ptr.as_ref())? };
+    let new_array = unsafe { import(array_ffi, &schema_ffi)? };
 
     // which is equal to the exported array
     assert_eq!(array.as_ref(), new_array.as_ref());
diff --git a/src/buffer/bytes.rs b/src/buffer/bytes.rs
index 3d573da5fe..67ad7bed13 100644
--- a/src/buffer/bytes.rs
+++ b/src/buffer/bytes.rs
@@ -1,14 +1,14 @@
 use std::mem::ManuallyDrop;
 use std::ops::{Deref, DerefMut};
 use std::panic::RefUnwindSafe;
-use std::{ptr::NonNull, sync::Arc};
+use std::ptr::NonNull;
 
 /// Mode of deallocating memory regions
 enum Allocation {
     /// Native allocation
     Native,
     // A foreign allocator and its ref count
-    Foreign(Arc<dyn RefUnwindSafe + Send + Sync>),
+    Foreign(Box<dyn RefUnwindSafe + Send + Sync>),
 }
 
 /// A continuous memory region that may be allocated externally.
@@ -35,7 +35,7 @@ impl<T> Bytes<T> {
     pub unsafe fn from_owned(
         ptr: std::ptr::NonNull<T>,
         len: usize,
-        owner: Arc<dyn RefUnwindSafe + Send + Sync>,
+        owner: Box<dyn RefUnwindSafe + Send + Sync>,
     ) -> Self {
         // This line is technically outside the assumptions of `Vec::from_raw_parts`, since
         // `ptr` was not allocated by `Vec`. However, one of the invariants of this struct
diff --git a/src/ffi/array.rs b/src/ffi/array.rs
index 700e83e2e0..6bce816255 100644
--- a/src/ffi/array.rs
+++ b/src/ffi/array.rs
@@ -1,5 +1,6 @@
 //! Contains functionality to load an ArrayData from the C Data Interface
-use std::{ptr::NonNull, sync::Arc};
+use std::ptr::NonNull;
+use std::sync::Arc;
 
 use crate::{
     array::*,
@@ -178,7 +179,7 @@ impl ArrowArray {
 unsafe fn create_buffer<T: NativeType>(
     array: &ArrowArray,
     data_type: &DataType,
-    owner: Arc<InternalArrowArray>,
+    owner: Box<InternalArrowArray>,
     index: usize,
 ) -> Result<Buffer<T>> {
     if array.buffers.is_null() {
@@ -209,7 +210,7 @@ unsafe fn create_buffer<T: NativeType>(
 /// This function assumes that `ceil(self.length * bits, 8)` is the size of the buffer
 unsafe fn create_bitmap(
     array: &ArrowArray,
-    owner: Arc<InternalArrowArray>,
+    owner: Box<InternalArrowArray>,
     index: usize,
 ) -> Result<Bitmap> {
     if array.buffers.is_null() {
@@ -310,7 +311,7 @@ fn buffer_len(array: &ArrowArray, data_type: &DataType, i: usize) -> Result<usiz
 fn create_child(
     array: &ArrowArray,
     field: &DataType,
-    parent: Arc<InternalArrowArray>,
+    parent: Box<InternalArrowArray>,
     index: usize,
 ) -> Result<ArrowArrayChild<'static>> {
     let data_type = get_child(field, index)?;
@@ -328,7 +329,7 @@ fn create_child(
 fn create_dictionary(
     array: &ArrowArray,
     data_type: &DataType,
-    parent: Arc<InternalArrowArray>,
+    parent: Box<InternalArrowArray>,
 ) -> Result<Option<ArrowArrayChild<'static>>> {
     if let DataType::Dictionary(_, values, _) = data_type {
         let data_type = values.as_ref().clone();
@@ -341,8 +342,8 @@ fn create_dictionary(
 }
 
 pub trait ArrowArrayRef: std::fmt::Debug {
-    fn owner(&self) -> Arc<InternalArrowArray> {
-        self.parent().clone()
+    fn owner(&self) -> Box<InternalArrowArray> {
+        (*self.parent()).clone()
     }
 
     /// returns the null bit buffer.
@@ -385,7 +386,7 @@ pub trait ArrowArrayRef: std::fmt::Debug {
 
     fn n_buffers(&self) -> usize;
 
-    fn parent(&self) -> &Arc<InternalArrowArray>;
+    fn parent(&self) -> &Box<InternalArrowArray>;
     fn array(&self) -> &ArrowArray;
     fn data_type(&self) -> &DataType;
 }
@@ -409,25 +410,30 @@ pub trait ArrowArrayRef: std::fmt::Debug {
 /// calling [ArrowArray::release] and [ArrowSchema::release] accordingly.
 ///
 /// Furthermore, this struct assumes that the incoming data agrees with the C data interface.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct InternalArrowArray {
-    array: Box<ArrowArray>,
-    data_type: DataType,
+    // Arc is used for sharability since this is immutable
+    array: Arc<ArrowArray>,
+    // Arced to reduce cost of cloning
+    data_type: Arc<DataType>,
 }
 
 impl InternalArrowArray {
-    pub fn new(array: Box<ArrowArray>, data_type: DataType) -> Self {
-        Self { array, data_type }
+    pub fn new(array: ArrowArray, data_type: DataType) -> Self {
+        Self {
+            array: Arc::new(array),
+            data_type: Arc::new(data_type),
+        }
     }
 }
 
-impl ArrowArrayRef for Arc<InternalArrowArray> {
+impl ArrowArrayRef for Box<InternalArrowArray> {
     /// the data_type as declared in the schema
     fn data_type(&self) -> &DataType {
         &self.data_type
     }
 
-    fn parent(&self) -> &Arc<InternalArrowArray> {
+    fn parent(&self) -> &Box<InternalArrowArray> {
         self
     }
 
@@ -444,7 +450,7 @@ impl ArrowArrayRef for Arc<InternalArrowArray> {
 pub struct ArrowArrayChild<'a> {
     array: &'a ArrowArray,
     data_type: DataType,
-    parent: Arc<InternalArrowArray>,
+    parent: Box<InternalArrowArray>,
 }
 
 impl<'a> ArrowArrayRef for ArrowArrayChild<'a> {
@@ -453,7 +459,7 @@ impl<'a> ArrowArrayRef for ArrowArrayChild<'a> {
         &self.data_type
     }
 
-    fn parent(&self) -> &Arc<InternalArrowArray> {
+    fn parent(&self) -> &Box<InternalArrowArray> {
         &self.parent
     }
 
@@ -470,7 +476,7 @@ impl<'a> ArrowArrayChild<'a> {
     fn from_raw(
         array: &'a ArrowArray,
         data_type: DataType,
-        parent: Arc<InternalArrowArray>,
+        parent: Box<InternalArrowArray>,
     ) -> Self {
         Self {
             array,
diff --git a/src/ffi/mod.rs b/src/ffi/mod.rs
index 4a0c82a055..10fc2fb994 100644
--- a/src/ffi/mod.rs
+++ b/src/ffi/mod.rs
@@ -9,8 +9,6 @@ mod stream;
 pub(crate) use array::try_from;
 pub(crate) use array::{ArrowArrayRef, InternalArrowArray};
 
-use std::sync::Arc;
-
 use crate::array::Array;
 use crate::datatypes::{DataType, Field};
 use crate::error::Result;
@@ -21,25 +19,19 @@ pub use generated::{ArrowArray, ArrowArrayStream, ArrowSchema};
 pub use stream::{export_iterator, ArrowArrayStreamReader};
 
 /// Exports an [`Box<dyn Array>`] to the C data interface.
-/// # Safety
-/// The pointer `ptr` must be allocated and valid
-pub unsafe fn export_array_to_c(array: Box<dyn Array>, ptr: *mut ArrowArray) {
-    let array = bridge::align_to_c_data_interface(array);
-
-    std::ptr::write_unaligned(ptr, ArrowArray::new(array));
+pub fn export_array_to_c(array: Box<dyn Array>) -> ArrowArray {
+    ArrowArray::new(bridge::align_to_c_data_interface(array))
 }
 
 /// Exports a [`Field`] to the C data interface.
-/// # Safety
-/// The pointer `ptr` must be allocated and valid
-pub unsafe fn export_field_to_c(field: &Field, ptr: *mut ArrowSchema) {
-    std::ptr::write_unaligned(ptr, ArrowSchema::new(field));
+pub fn export_field_to_c(field: &Field) -> ArrowSchema {
+    ArrowSchema::new(field)
 }
 
 /// Imports a [`Field`] from the C data interface.
 /// # Safety
 /// This function is intrinsically `unsafe` and relies on a [`ArrowSchema`]
-/// valid according to the [C data interface](https://arrow.apache.org/docs/format/CDataInterface.html) (FFI).
+/// being valid according to the [C data interface](https://arrow.apache.org/docs/format/CDataInterface.html) (FFI).
 pub unsafe fn import_field_from_c(field: &ArrowSchema) -> Result<Field> {
     to_field(field)
 }
@@ -47,10 +39,10 @@ pub unsafe fn import_field_from_c(field: &ArrowSchema) -> Result<Field> {
 /// Imports an [`Array`] from the C data interface.
 /// # Safety
 /// This function is intrinsically `unsafe` and relies on a [`ArrowArray`]
-/// valid according to the [C data interface](https://arrow.apache.org/docs/format/CDataInterface.html) (FFI).
+/// being valid according to the [C data interface](https://arrow.apache.org/docs/format/CDataInterface.html) (FFI).
 pub unsafe fn import_array_from_c(
-    array: Box<ArrowArray>,
+    array: ArrowArray,
     data_type: DataType,
 ) -> Result<Box<dyn Array>> {
-    try_from(Arc::new(InternalArrowArray::new(array, data_type)))
+    try_from(Box::new(InternalArrowArray::new(array, data_type)))
 }
diff --git a/src/ffi/stream.rs b/src/ffi/stream.rs
index ab128e4626..5e0dabe012 100644
--- a/src/ffi/stream.rs
+++ b/src/ffi/stream.rs
@@ -61,8 +61,6 @@ impl ArrowArrayStreamReader {
     /// * The `ArrowArrayStream` fulfills the invariants of the C stream interface
     /// * The schema `get_schema` produces fulfills the C data interface
     pub unsafe fn try_new(mut iter: Box<ArrowArrayStream>) -> Result<Self, Error> {
-        let mut field = Box::new(ArrowSchema::empty());
-
         if iter.get_next.is_none() {
             return Err(Error::OutOfSpec(
                 "The C stream MUST contain a non-null get_next".to_string(),
@@ -75,8 +73,9 @@ impl ArrowArrayStreamReader {
             ));
         };
 
+        let mut field = ArrowSchema::empty();
         let status = if let Some(f) = iter.get_schema {
-            unsafe { (f)(&mut *iter, &mut *field) }
+            unsafe { (f)(&mut *iter, &mut field) }
         } else {
             return Err(Error::OutOfSpec(
                 "The C stream MUST contain a non-null get_schema".to_string(),
@@ -106,8 +105,8 @@ impl ArrowArrayStreamReader {
     /// Calling this iterator's `next` assumes that the [`ArrowArrayStream`] produces arrow arrays
     /// that fulfill the C data interface
     pub unsafe fn next(&mut self) -> Option<Result<Box<dyn Array>, Error>> {
-        let mut array = Box::new(ArrowArray::empty());
-        let status = unsafe { (self.iter.get_next.unwrap())(&mut *self.iter, &mut *array) };
+        let mut array = ArrowArray::empty();
+        let status = unsafe { (self.iter.get_next.unwrap())(&mut *self.iter, &mut array) };
 
         if status != 0 {
             return Some(Err(unsafe { handle_error(&mut self.iter) }));
@@ -145,7 +144,8 @@ unsafe extern "C" fn get_next(iter: *mut ArrowArrayStream, array: *mut ArrowArra
                 return 2001; // custom application specific error (since this is never a result of this interface)
             }
 
-            export_array_to_c(item, array);
+            std::ptr::write(array, export_array_to_c(item));
+
             private.error = None;
             0
         }
@@ -168,7 +168,7 @@ unsafe extern "C" fn get_schema(iter: *mut ArrowArrayStream, schema: *mut ArrowS
     }
     let private = &mut *((*iter).private_data as *mut PrivateData);
 
-    export_field_to_c(&private.field, schema);
+    std::ptr::write(schema, export_field_to_c(&private.field));
     0
 }
 
@@ -195,20 +195,17 @@ unsafe extern "C" fn release(iter: *mut ArrowArrayStream) {
 }
 
 /// Exports an iterator to the [C stream interface](https://arrow.apache.org/docs/format/CStreamInterface.html)
-/// # Safety
-/// The pointer `consumer` must be allocated
-pub unsafe fn export_iterator(
+pub fn export_iterator(
     iter: Box<dyn Iterator<Item = Result<Box<dyn Array>, Error>>>,
     field: Field,
-    consumer: *mut ArrowArrayStream,
-) {
+) -> ArrowArrayStream {
     let private_data = Box::new(PrivateData {
         iter,
         field,
         error: None,
     });
 
-    *consumer = ArrowArrayStream {
+    ArrowArrayStream {
         get_schema: Some(get_schema),
         get_next: Some(get_next),
         get_last_error: Some(get_last_error),
diff --git a/src/io/parquet/read/deserialize/README.md b/src/io/parquet/read/deserialize/README.md
new file mode 100644
index 0000000000..971ed42478
--- /dev/null
+++ b/src/io/parquet/read/deserialize/README.md
@@ -0,0 +1,70 @@
+# Design
+
+## Non-nested types
+
+Let's start with the design used for non-nested arrays. The (private) entry point of this
+module for non-nested arrays is `simple::page_iter_to_arrays`.
+
+This function expects
+
+* a (fallible) streaming iterator of decompressed and encoded pages, `DataPages`
+* the source (parquet) column type, including its logical information
+* the target (arrow) `DataType`
+* the chunk size
+
+and returns an iterator of `Array`, `ArrayIter`.
+
+This design is shared among _all_ `(parquet, arrow)` implemented tuples. Their main
+difference is how they are deserialized, which depends on the source and target types.
+
+When the array iterator is pulled the first time, the following happens:
+* a page from `DataPages` is pulled
+* a `PageState<'a>` is built from the page
+* the `PageState` is consumed into a mutable array:
+    * if `chunk_size` is larger than the number of rows in the page, the mutable array state is preserved and a new page is pulled and the process repeated until we fill a chunk.
+    * if `chunk_size` is smaller than the number of rows in the page, the mutable array state
+      is returned and the remaining of the page is consumed into multiple mutable arrays of length `chunk_size` into a FIFO queue.
+
+Subsequent pulls of arrays will first try to pull from the FIFO queue. Once the queue is empty, the
+a new page is pulled.
+
+### `PageState`
+
+As mentioned above, the iterator leverages the idea that we attach a state to a page. Recall
+that a page is essentially `[header][data]`. The `data` part contains encoded 
+`[rep levels][def levels][non-null values]`. Some pages have an associated dictionary page,
+in which case the `non-null values` represent the indices.
+
+Irrespectively of the physical type, the main idea is to split the page in two iterators:
+
+* An iterator over `def levels`
+* An iterator over `non-null values`
+
+and progress the iterators as needed. In particular, for non-nested types, `def levels` is 
+a bitmap with the same representation as Arrow, in which case the validity is extended directly.
+
+The `non-null values` are "expanded" by filling null values with the default value of each physical
+type.
+
+## Nested types
+
+For nested type with N+1 levels (1 is the primitive), we need to build the nest information of each
+N levels + the non-nested Arrow array.
+
+This is done by first transversing the parquet types and using it to initialize, per chunk, the N levels.
+
+The per-chunk execution is then similar but `chunk_size` only drives the number of retrieved
+rows from the outermost parquet group (the field). Each of these pulls knows how many items need
+to be pulled from the inner groups, all the way to the primitive type. This works because
+in parquet a row cannot be split between two pages and thus each page is guaranteed
+to contain a full row.
+
+The `PageState` of nested types is composed by 4 iterators:
+
+* A (zipped) iterator over `rep levels` and `def levels`
+* An iterator over `def levels`
+* An iterator over `non-null values`
+
+The idea is that an iterator of `rep, def` contain all the information to decode the 
+nesting structure of an arrow array. The other two iterators are equivalent to the non-nested
+types with the exception that `def levels` are no equivalent to arrow bitmaps.
diff --git a/src/io/parquet/read/deserialize/simple.rs b/src/io/parquet/read/deserialize/simple.rs
index 8a8ef653da..aaab8aee9f 100644
--- a/src/io/parquet/read/deserialize/simple.rs
+++ b/src/io/parquet/read/deserialize/simple.rs
@@ -55,7 +55,7 @@ where
 }
 
 /// An iterator adapter that maps an iterator of DataPages into an iterator of Arrays
-/// of [`DataType`] `data_type` and `chunk_size`.
+/// of [`DataType`] `data_type` and length `chunk_size`.
 pub fn page_iter_to_arrays<'a, I: 'a + DataPages>(
     pages: I,
     type_: &PrimitiveType,
diff --git a/src/io/parquet/write/pages.rs b/src/io/parquet/write/pages.rs
index 4c0c0c2342..28f45a06fd 100644
--- a/src/io/parquet/write/pages.rs
+++ b/src/io/parquet/write/pages.rs
@@ -172,7 +172,7 @@ fn to_leafs_recursive<'a>(array: &'a dyn Array, leafs: &mut Vec<&'a dyn Array>)
         }
         Null | Boolean | Primitive(_) | Binary | FixedSizeBinary | LargeBinary | Utf8
         | LargeUtf8 | Dictionary(_) => leafs.push(array),
-        _ => todo!(),
+        other => todo!("Writing {:?} to parquet not yet implemented", other),
     }
 }
 


Test patch:
diff --git a/arrow-parquet-integration-testing/main.py b/arrow-parquet-integration-testing/main.py
index 545a6e08b5..42e19e81b8 100644
--- a/arrow-parquet-integration-testing/main.py
+++ b/arrow-parquet-integration-testing/main.py
@@ -49,11 +49,8 @@ non_native_types = [
     pyarrow.date64(),
     pyarrow.time32("s"),
     pyarrow.timestamp("s"),
+    # the issue here is the second, not the tz
     pyarrow.timestamp("s", tz="UTC"),
-    pyarrow.duration("s"),
-    pyarrow.duration("ms"),
-    pyarrow.duration("us"),
-    pyarrow.duration("ns"),
 ]
 
 
diff --git a/arrow-parquet-integration-testing/src/main.rs b/arrow-parquet-integration-testing/src/main.rs
index c7e9a03431..f62fba5e09 100644
--- a/arrow-parquet-integration-testing/src/main.rs
+++ b/arrow-parquet-integration-testing/src/main.rs
@@ -11,8 +11,8 @@ use arrow2::{
         json_integration::read,
         json_integration::ArrowJson,
         parquet::write::{
-            CompressionOptions as ParquetCompression, Encoding, FileWriter, RowGroupIterator,
-            Version as ParquetVersion, WriteOptions,
+            transverse, CompressionOptions as ParquetCompression, Encoding, FileWriter,
+            RowGroupIterator, Version as ParquetVersion, WriteOptions,
         },
     },
 };
@@ -174,16 +174,18 @@ fn main() -> Result<()> {
     let encodings = schema
         .fields
         .iter()
-        .map(|x| match x.data_type() {
-            DataType::Dictionary(..) => vec![Encoding::RleDictionary],
-            DataType::Utf8 | DataType::LargeUtf8 => {
-                vec![if args.encoding_utf8 == EncodingScheme::Delta {
-                    Encoding::DeltaLengthByteArray
-                } else {
-                    Encoding::Plain
-                }]
-            }
-            _ => vec![Encoding::Plain],
+        .map(|f| {
+            transverse(&f.data_type, |dt| match dt {
+                DataType::Dictionary(..) => Encoding::RleDictionary,
+                DataType::Utf8 | DataType::LargeUtf8 => {
+                    if args.encoding_utf8 == EncodingScheme::Delta {
+                        Encoding::DeltaLengthByteArray
+                    } else {
+                        Encoding::Plain
+                    }
+                }
+                _ => Encoding::Plain,
+            })
         })
         .collect();
 
diff --git a/arrow-pyarrow-integration-testing/src/c_stream.rs b/arrow-pyarrow-integration-testing/src/c_stream.rs
index 8998f4812e..4f10f5cb4c 100644
--- a/arrow-pyarrow-integration-testing/src/c_stream.rs
+++ b/arrow-pyarrow-integration-testing/src/c_stream.rs
@@ -23,7 +23,7 @@ pub fn to_rust_iterator(ob: PyObject, py: Python) -> PyResult<Vec<PyObject>> {
 
     let mut arrays = vec![];
     while let Some(array) = unsafe { iter.next() } {
-        let py_array = to_py_array(array.unwrap().into(), py)?;
+        let py_array = to_py_array(array.map_err(PyO3Error::from)?, py)?;
         arrays.push(py_array)
     }
     Ok(arrays)
@@ -38,16 +38,16 @@ pub fn from_rust_iterator(py: Python) -> PyResult<PyObject> {
         None,
     )
     .boxed();
+
     // and a field with its datatype
     let field = Field::new("a", array.data_type().clone(), true);
 
     // create an iterator of arrays
     let arrays = vec![array.clone(), array.clone(), array];
-    let iter = Box::new(arrays.clone().into_iter().map(Ok)) as _;
+    let iter = Box::new(arrays.into_iter().map(Ok)) as _;
 
     // create an [`ArrowArrayStream`] based on this iterator and field
-    let mut stream = Box::new(ffi::ArrowArrayStream::empty());
-    unsafe { ffi::export_iterator(iter, field, &mut *stream) };
+    let stream = Box::new(ffi::export_iterator(iter, field));
 
     // call pyarrow's interface to read this stream
     let pa = py.import("pyarrow.ipc")?;
@@ -55,7 +55,6 @@ pub fn from_rust_iterator(py: Python) -> PyResult<PyObject> {
         "_import_from_c",
         ((&*stream as *const ffi::ArrowArrayStream) as Py_uintptr_t,),
     )?;
-    Box::leak(stream); // this is not ideal => the struct should be allocated by pyarrow
 
     Ok(py_stream.to_object(py))
 }
diff --git a/arrow-pyarrow-integration-testing/src/lib.rs b/arrow-pyarrow-integration-testing/src/lib.rs
index ad6a16bbad..ddfe800599 100644
--- a/arrow-pyarrow-integration-testing/src/lib.rs
+++ b/arrow-pyarrow-integration-testing/src/lib.rs
@@ -67,22 +67,21 @@ fn to_rust_array(ob: PyObject, py: Python) -> PyResult<Box<dyn Array>> {
 
     let field = unsafe { ffi::import_field_from_c(schema.as_ref()).map_err(PyO3Error::from)? };
     let array =
-        unsafe { ffi::import_array_from_c(array, field.data_type).map_err(PyO3Error::from)? };
+        unsafe { ffi::import_array_from_c(*array, field.data_type).map_err(PyO3Error::from)? };
 
-    Ok(array.into())
+    Ok(array)
 }
 
 fn to_py_array(array: Box<dyn Array>, py: Python) -> PyResult<PyObject> {
-    let array_ptr = Box::new(ffi::ArrowArray::empty());
-    let schema_ptr = Box::new(ffi::ArrowSchema::empty());
+    let schema = Box::new(ffi::export_field_to_c(&Field::new(
+        "",
+        array.data_type().clone(),
+        true,
+    )));
+    let array = Box::new(ffi::export_array_to_c(array));
 
-    let array_ptr = Box::into_raw(array_ptr);
-    let schema_ptr = Box::into_raw(schema_ptr);
-
-    unsafe {
-        ffi::export_field_to_c(&Field::new("", array.data_type().clone(), true), schema_ptr);
-        ffi::export_array_to_c(array, array_ptr);
-    };
+    let schema_ptr: *const arrow2::ffi::ArrowSchema = &*schema;
+    let array_ptr: *const arrow2::ffi::ArrowArray = &*array;
 
     let pa = py.import("pyarrow")?;
 
@@ -91,11 +90,6 @@ fn to_py_array(array: Box<dyn Array>, py: Python) -> PyResult<PyObject> {
         (array_ptr as Py_uintptr_t, schema_ptr as Py_uintptr_t),
     )?;
 
-    unsafe {
-        Box::from_raw(array_ptr);
-        Box::from_raw(schema_ptr);
-    };
-
     Ok(array.to_object(py))
 }
 
@@ -115,12 +109,8 @@ fn to_rust_field(ob: PyObject, py: Python) -> PyResult<Field> {
 }
 
 fn to_py_field(field: &Field, py: Python) -> PyResult<PyObject> {
-    let schema_ptr = Box::new(ffi::ArrowSchema::empty());
-    let schema_ptr = Box::into_raw(schema_ptr);
-
-    unsafe {
-        ffi::export_field_to_c(field, schema_ptr);
-    };
+    let schema = Box::new(ffi::export_field_to_c(field));
+    let schema_ptr: *const arrow2::ffi::ArrowSchema = &*schema;
 
     let pa = py.import("pyarrow")?;
 
@@ -128,8 +118,6 @@ fn to_py_field(field: &Field, py: Python) -> PyResult<PyObject> {
         .getattr("Field")?
         .call_method1("_import_from_c", (schema_ptr as Py_uintptr_t,))?;
 
-    unsafe { Box::from_raw(schema_ptr) };
-
     Ok(array.to_object(py))
 }
 
diff --git a/tests/it/array/utf8/to_mutable.rs b/tests/it/array/utf8/to_mutable.rs
index 12f52151c6..e9956feffd 100644
--- a/tests/it/array/utf8/to_mutable.rs
+++ b/tests/it/array/utf8/to_mutable.rs
@@ -1,14 +1,67 @@
-use arrow2::array::Utf8Array;
+use arrow2::{array::Utf8Array, bitmap::Bitmap, buffer::Buffer, datatypes::DataType};
 
 #[test]
-#[allow(clippy::redundant_clone)]
-fn array_to_mutable() {
+fn not_shared() {
     let array = Utf8Array::<i32>::from(&[Some("hello"), Some(" "), None]);
-    let mutable = array.into_mut().unwrap_right();
+    assert!(array.into_mut().is_right());
+}
+
+#[test]
+#[allow(clippy::redundant_clone)]
+fn shared_validity() {
+    let validity = Bitmap::from([true]);
+    let array = Utf8Array::<i32>::new(
+        DataType::Utf8,
+        vec![0, 1].into(),
+        b"a".to_vec().into(),
+        Some(validity.clone()),
+    );
+    assert!(array.into_mut().is_left())
+}
 
-    let array: Utf8Array<i32> = mutable.into();
-    let array2 = array.clone();
-    let maybe_mut = array2.into_mut();
-    // the ref count is 2 we should not get a mutable.
-    assert!(maybe_mut.is_left())
+#[test]
+#[allow(clippy::redundant_clone)]
+fn shared_values() {
+    let values: Buffer<u8> = b"a".to_vec().into();
+    let array = Utf8Array::<i32>::new(
+        DataType::Utf8,
+        vec![0, 1].into(),
+        values.clone(),
+        Some(Bitmap::from([true])),
+    );
+    assert!(array.into_mut().is_left())
+}
+
+#[test]
+#[allow(clippy::redundant_clone)]
+fn shared_offsets_values() {
+    let offsets: Buffer<i32> = vec![0, 1].into();
+    let values: Buffer<u8> = b"a".to_vec().into();
+    let array = Utf8Array::<i32>::new(
+        DataType::Utf8,
+        offsets.clone(),
+        values.clone(),
+        Some(Bitmap::from([true])),
+    );
+    assert!(array.into_mut().is_left())
+}
+
+#[test]
+#[allow(clippy::redundant_clone)]
+fn shared_offsets() {
+    let offsets: Buffer<i32> = vec![0, 1].into();
+    let array = Utf8Array::<i32>::new(
+        DataType::Utf8,
+        offsets.clone(),
+        b"a".to_vec().into(),
+        Some(Bitmap::from([true])),
+    );
+    assert!(array.into_mut().is_left())
+}
+
+#[test]
+#[allow(clippy::redundant_clone)]
+fn shared_all() {
+    let array = Utf8Array::<i32>::from(&[Some("hello"), Some(" "), None]);
+    assert!(array.clone().into_mut().is_left())
 }
diff --git a/tests/it/ffi/data.rs b/tests/it/ffi/data.rs
index 29a0122384..337326b631 100644
--- a/tests/it/ffi/data.rs
+++ b/tests/it/ffi/data.rs
@@ -7,24 +7,14 @@ use std::collections::BTreeMap;
 fn _test_round_trip(array: Box<dyn Array>, expected: Box<dyn Array>) -> Result<()> {
     let field = Field::new("a", array.data_type().clone(), true);
 
-    let array_ptr = Box::new(ffi::ArrowArray::empty());
-    let schema_ptr = Box::new(ffi::ArrowSchema::empty());
-
-    let array_ptr = Box::into_raw(array_ptr);
-    let schema_ptr = Box::into_raw(schema_ptr);
-
-    unsafe {
-        ffi::export_array_to_c(array, array_ptr);
-        ffi::export_field_to_c(&field, schema_ptr);
-    }
-
-    let array_ptr = unsafe { Box::from_raw(array_ptr) };
-    let schema_ptr = unsafe { Box::from_raw(schema_ptr) };
+    // export array and corresponding data_type
+    let array_ffi = ffi::export_array_to_c(array);
+    let schema_ffi = ffi::export_field_to_c(&field);
 
     // import references
-    let result_field = unsafe { ffi::import_field_from_c(schema_ptr.as_ref())? };
+    let result_field = unsafe { ffi::import_field_from_c(&schema_ffi)? };
     let result_array =
-        unsafe { ffi::import_array_from_c(array_ptr, result_field.data_type.clone())? };
+        unsafe { ffi::import_array_from_c(array_ffi, result_field.data_type.clone())? };
 
     assert_eq!(&result_array, &expected);
     assert_eq!(result_field, field);
@@ -41,16 +31,9 @@ fn test_round_trip(expected: impl Array + Clone + 'static) -> Result<()> {
 }
 
 fn test_round_trip_schema(field: Field) -> Result<()> {
-    // create a `InternalArrowArray` from the data.
-    let schema_ptr = Box::new(ffi::ArrowSchema::empty());
-
-    let schema_ptr = Box::into_raw(schema_ptr);
-
-    unsafe { ffi::export_field_to_c(&field, schema_ptr) };
-
-    let schema_ptr = unsafe { Box::from_raw(schema_ptr) };
+    let schema_ffi = ffi::export_field_to_c(&field);
 
-    let result = unsafe { ffi::import_field_from_c(schema_ptr.as_ref())? };
+    let result = unsafe { ffi::import_field_from_c(&schema_ffi)? };
 
     assert_eq!(result, field);
     Ok(())
diff --git a/tests/it/ffi/stream.rs b/tests/it/ffi/stream.rs
index 944da6e9b4..44d0e1e7cc 100644
--- a/tests/it/ffi/stream.rs
+++ b/tests/it/ffi/stream.rs
@@ -8,8 +8,9 @@ fn _test_round_trip(arrays: Vec<Box<dyn Array>>) -> Result<()> {
 
     let mut stream = Box::new(ffi::ArrowArrayStream::empty());
 
-    unsafe { ffi::export_iterator(iter, field.clone(), &mut *stream) }
+    *stream = ffi::export_iterator(iter, field.clone());
 
+    // import
     let mut stream = unsafe { ffi::ArrowArrayStreamReader::try_new(stream)? };
 
     let mut produced_arrays: Vec<Box<dyn Array>> = vec![];