Repository: jorgecarleitao/arrow2
Pull request #1141 (2022-07-07 02:04:24)

Moved `Bytes` to own crate
This PR proposes that we move the `Bytes` API to another crate so others can use it without having to copy paste. It also separates concerns between this crate (arrow) and a mechanism to use share foreign arrays with Rust still compatible with mutable APIs.

Proposed home: https://github.com/DataEngineeringLabs/foreign_vec

codecov[bot]: # [Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1141?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) Report
> Merging [#1141](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1141?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (0fc0182) into [main](https://codecov.io/gh/jorgecarleitao/arrow2/commit/98e49133b2e56e51e30335830485b3cf768eb5a2?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) (98e4913) will **decrease** coverage by `0.01%`.
> The diff coverage is `90.90%`.

```diff
@@            Coverage Diff             @@
##             main    #1141      +/-   ##
==========================================
- Coverage   83.59%   83.58%   -0.02%     
==========================================
  Files         366      365       -1     
  Lines       35901    35851      -50     
==========================================
- Hits        30013    29966      -47     
+ Misses       5888     5885       -3     
```


| [Impacted Files](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1141?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | Coverage Δ | |
|---|---|---|
| [src/array/ffi.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1141/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2ZmaS5ycw==) | `86.95% <ø> (ø)` | |
| [src/array/map/ffi.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1141/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L21hcC9mZmkucnM=) | `0.00% <0.00%> (ø)` | |
| [src/array/null.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1141/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L251bGwucnM=) | `45.67% <0.00%> (ø)` | |
| [src/array/union/ffi.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1141/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L3VuaW9uL2ZmaS5ycw==) | `0.00% <ø> (ø)` | |
| [src/array/binary/ffi.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1141/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2JpbmFyeS9mZmkucnM=) | `97.50% <100.00%> (ø)` | |
| [src/array/boolean/ffi.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1141/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2Jvb2xlYW4vZmZpLnJz) | `97.14% <100.00%> (ø)` | |
| [src/array/dictionary/ffi.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1141/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2RpY3Rpb25hcnkvZmZpLnJz) | `70.83% <100.00%> (ø)` | |
| [src/array/fixed\_size\_binary/ffi.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1141/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2ZpeGVkX3NpemVfYmluYXJ5L2ZmaS5ycw==) | `97.22% <100.00%> (ø)` | |
| [src/array/fixed\_size\_list/ffi.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1141/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2ZpeGVkX3NpemVfbGlzdC9mZmkucnM=) | `87.50% <100.00%> (ø)` | |
| [src/array/list/ffi.rs](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1141/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao#diff-c3JjL2FycmF5L2xpc3QvZmZpLnJz) | `97.56% <100.00%> (ø)` | |
| ... and [9 more](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1141/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1141?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/1141?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Jorge+Leitao). Last update [98e4913...0fc0182](https://codecov.io/gh/jorgecarleitao/arrow2/pull/1141?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).

jorgecarleitao: cc @jhorstmann @ritchie46 , since both also contributed to this component (either here or in arrow-rs).

Patch:
diff --git a/Cargo.toml b/Cargo.toml
index 4bc21d5948..1aada784f4 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -15,6 +15,7 @@ name = "arrow2"
 bench = false
 
 [dependencies]
+foreign_vec = "0.1.0"
 either = "1.6"
 num-traits = "0.2"
 dyn-clone = "1"
@@ -64,7 +65,7 @@ base64 = { version = "0.13.0", optional = true }
 futures = { version = "0.3", optional = true }
 
 # for faster hashing
-ahash = { version = "0.7", optional = true }
+ahash = { version = "0.7" }
 
 # parquet support
 parquet2 = { version = "0.14.0", optional = true, default_features = false }
@@ -175,7 +176,7 @@ compute_comparison = ["compute_take", "compute_boolean"]
 compute_concatenate = []
 compute_contains = []
 compute_filter = []
-compute_hash = ["multiversion", "ahash"]
+compute_hash = ["multiversion"]
 compute_if_then_else = []
 compute_length = []
 compute_like = ["regex"]
diff --git a/examples/avro_kafka.rs b/examples/avro_kafka.rs
new file mode 100644
index 0000000000..0ee461a5b7
--- /dev/null
+++ b/examples/avro_kafka.rs
@@ -0,0 +1,54 @@
+use arrow2::{
+    datatypes::{DataType, Field},
+    error::Error,
+    io::avro,
+};
+
+fn read_schema_id<R: std::io::Read>(reader: &mut R) -> Result<i32, Error> {
+    let mut header = [0; 5];
+    reader.read_exact(&mut header)?;
+
+    if header[0] != 0 {
+        return Err(Error::ExternalFormat(
+            "Avro requires the first byte to be a zero".to_string(),
+        ));
+    }
+    Ok(i32::from_be_bytes(header[1..].try_into().unwrap()))
+}
+
+fn read_block<R: std::io::Read>(reader: &mut R, block: &mut avro::Block) -> Result<(), Error> {
+    // (we could lump multiple records together by extending the block instead of clearing)
+    block.data.clear();
+    reader.read_to_end(&mut block.data)?;
+    block.number_of_rows = 1;
+    Ok(())
+}
+
+fn main() -> Result<(), Error> {
+    // say we received the event from Kafka
+    let data = &[0, 0, 0, 0, 1, 6, 115, 99, 105, 6, 109, 97, 115];
+    let mut stream = std::io::Cursor::new(data);
+
+    // we can fetch its schema_id from the registry via:
+    let _schema_id = read_schema_id(&mut stream)?;
+
+    // say that from the registry we concluded that this schema has fields
+    let avro_fields = vec![
+        avro_schema::Schema::String(None),
+        avro_schema::Schema::String(None),
+    ];
+    // (which we map to arrow fields as)
+    let fields = vec![
+        Field::new("first_name", DataType::Utf8, false),
+        Field::new("last_name", DataType::Utf8, false),
+    ];
+
+    // the below allow us to read it to arrow (a chunk of a single element)
+    let mut block = avro::Block::default();
+    read_block(&mut stream, &mut block)?;
+
+    let chunk = avro::read::deserialize(&block, &fields, &avro_fields, &[true, true])?;
+
+    println!("{chunk:?}");
+    Ok(())
+}
diff --git a/parquet_integration/write_parquet.py b/parquet_integration/write_parquet.py
index 3fa56c3579..a80e69b68a 100644
--- a/parquet_integration/write_parquet.py
+++ b/parquet_integration/write_parquet.py
@@ -227,6 +227,15 @@ def case_struct() -> Tuple[dict, pa.Schema, str]:
                 "struct_nullable",
                 pa.struct(struct_fields),
             ),
+            pa.field(
+                "struct_struct_nullable",
+                pa.struct(
+                    [
+                        ("f1", pa.struct(struct_fields)),
+                        ("f2", pa.bool_()),
+                    ]
+                ),
+            ),
         ]
     )
 
@@ -248,6 +257,11 @@ def case_struct() -> Tuple[dict, pa.Schema, str]:
                 names=["f1", "f2"],
             ),
             "struct_nullable": struct_nullable,
+            "struct_struct_nullable": pa.StructArray.from_arrays(
+                [struct, pa.array(boolean)],
+                names=["f1", "f2"],
+                mask=pa.array([False, False, True, False, False, False, False, False, False, False]),
+            ),
         },
         schema,
         f"struct_nullable_10.parquet",
@@ -257,15 +271,30 @@ def case_struct() -> Tuple[dict, pa.Schema, str]:
 def case_nested_edge():
     simple = [[0, 1]]
     null = [None]
+
+    struct_list_nullable = pa.StructArray.from_arrays(
+        [pa.array([["a", "b", None, "c"]])],
+        fields=[
+            ("f1", pa.list_(pa.utf8())),
+        ],
+    )
+
     fields = [
         pa.field("simple", pa.list_(pa.int64())),
         pa.field("null", pa.list_(pa.field("item", pa.int64(), True))),
+        pa.field(
+                "struct_list_nullable",
+                pa.struct([
+                    ("f1", pa.list_(pa.utf8())),
+                ]),
+            )
     ]
     schema = pa.schema(fields)
     return (
         {
             "simple": simple,
             "null": null,
+            "struct_list_nullable": struct_list_nullable,
         },
         schema,
         f"nested_edge_nullable_10.parquet",
diff --git a/src/array/binary/ffi.rs b/src/array/binary/ffi.rs
index c8a72615bc..2c6792237a 100644
--- a/src/array/binary/ffi.rs
+++ b/src/array/binary/ffi.rs
@@ -9,7 +9,7 @@ use crate::error::Result;
 use super::BinaryArray;
 
 unsafe impl<O: Offset> ToFfi for BinaryArray<O> {
-    fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
+    fn buffers(&self) -> Vec<Option<*const u8>> {
         vec![
             self.validity.as_ref().map(|x| x.as_ptr()),
             Some(self.offsets.as_ptr().cast::<u8>()),
diff --git a/src/array/boolean/ffi.rs b/src/array/boolean/ffi.rs
index da7482c11e..c2081c49a5 100644
--- a/src/array/boolean/ffi.rs
+++ b/src/array/boolean/ffi.rs
@@ -9,7 +9,7 @@ use crate::error::Result;
 use super::BooleanArray;
 
 unsafe impl ToFfi for BooleanArray {
-    fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
+    fn buffers(&self) -> Vec<Option<*const u8>> {
         vec![
             self.validity.as_ref().map(|x| x.as_ptr()),
             Some(self.values.as_ptr()),
diff --git a/src/array/dictionary/ffi.rs b/src/array/dictionary/ffi.rs
index 1e06f81e6e..fb84665276 100644
--- a/src/array/dictionary/ffi.rs
+++ b/src/array/dictionary/ffi.rs
@@ -7,7 +7,7 @@ use crate::{
 use super::{DictionaryArray, DictionaryKey};
 
 unsafe impl<K: DictionaryKey> ToFfi for DictionaryArray<K> {
-    fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
+    fn buffers(&self) -> Vec<Option<*const u8>> {
         self.keys.buffers()
     }
 
diff --git a/src/array/ffi.rs b/src/array/ffi.rs
index c879b7fa03..141cab327e 100644
--- a/src/array/ffi.rs
+++ b/src/array/ffi.rs
@@ -9,7 +9,7 @@ use crate::error::Result;
 /// Implementing this trait incorrect will lead to UB
 pub(crate) unsafe trait ToFfi {
     /// The pointers to the buffers.
-    fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>>;
+    fn buffers(&self) -> Vec<Option<*const u8>>;
 
     /// The children
     fn children(&self) -> Vec<Box<dyn Array>> {
@@ -47,7 +47,7 @@ macro_rules! ffi_dyn {
 
 type BuffersChildren = (
     usize,
-    Vec<Option<std::ptr::NonNull<u8>>>,
+    Vec<Option<*const u8>>,
     Vec<Box<dyn Array>>,
     Option<Box<dyn Array>>,
 );
diff --git a/src/array/fixed_size_binary/ffi.rs b/src/array/fixed_size_binary/ffi.rs
index 1bb07d322b..f1775e74d5 100644
--- a/src/array/fixed_size_binary/ffi.rs
+++ b/src/array/fixed_size_binary/ffi.rs
@@ -8,7 +8,7 @@ use crate::{
 use super::FixedSizeBinaryArray;
 
 unsafe impl ToFfi for FixedSizeBinaryArray {
-    fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
+    fn buffers(&self) -> Vec<Option<*const u8>> {
         vec![
             self.validity.as_ref().map(|x| x.as_ptr()),
             Some(self.values.as_ptr().cast::<u8>()),
diff --git a/src/array/fixed_size_list/ffi.rs b/src/array/fixed_size_list/ffi.rs
index 47367b180a..128496c295 100644
--- a/src/array/fixed_size_list/ffi.rs
+++ b/src/array/fixed_size_list/ffi.rs
@@ -9,7 +9,7 @@ use crate::{
 };
 
 unsafe impl ToFfi for FixedSizeListArray {
-    fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
+    fn buffers(&self) -> Vec<Option<*const u8>> {
         vec![self.validity.as_ref().map(|x| x.as_ptr())]
     }
 
diff --git a/src/array/growable/dictionary.rs b/src/array/growable/dictionary.rs
index 0b3c432525..930f4268f2 100644
--- a/src/array/growable/dictionary.rs
+++ b/src/array/growable/dictionary.rs
@@ -88,11 +88,22 @@ impl<'a, T: DictionaryKey> GrowableDictionary<'a, T> {
         let validity = std::mem::take(&mut self.key_validity);
         let key_values = std::mem::take(&mut self.key_values);
 
+        #[cfg(debug_assertions)]
+        {
+            crate::array::specification::check_indexes(&key_values, self.values.len()).unwrap();
+        }
         let keys =
-            PrimitiveArray::<T>::try_new(T::PRIMITIVE.into(), key_values.into(), validity.into())
-                .unwrap();
-
-        DictionaryArray::<T>::try_new(self.data_type.clone(), keys, self.values.clone()).unwrap()
+            PrimitiveArray::<T>::new(T::PRIMITIVE.into(), key_values.into(), validity.into());
+
+        // Safety - the invariant of this struct ensures that this is up-held
+        unsafe {
+            DictionaryArray::<T>::try_new_unchecked(
+                self.data_type.clone(),
+                keys,
+                self.values.clone(),
+            )
+            .unwrap()
+        }
     }
 }
 
@@ -141,12 +152,7 @@ impl<'a, T: DictionaryKey> Growable<'a> for GrowableDictionary<'a, T> {
 
 impl<'a, T: DictionaryKey> From<GrowableDictionary<'a, T>> for DictionaryArray<T> {
     #[inline]
-    fn from(val: GrowableDictionary<'a, T>) -> Self {
-        let data_type = T::PRIMITIVE.into();
-        let keys =
-            PrimitiveArray::<T>::try_new(data_type, val.key_values.into(), val.key_validity.into())
-                .unwrap();
-
-        DictionaryArray::<T>::try_new(val.data_type.clone(), keys, val.values).unwrap()
+    fn from(mut val: GrowableDictionary<'a, T>) -> Self {
+        val.to()
     }
 }
diff --git a/src/array/growable/list.rs b/src/array/growable/list.rs
index 41db9ae789..0e1e7ceb5f 100644
--- a/src/array/growable/list.rs
+++ b/src/array/growable/list.rs
@@ -104,12 +104,20 @@ impl<'a, O: Offset> GrowableList<'a, O> {
         let offsets = std::mem::take(&mut self.offsets);
         let values = self.values.as_box();
 
-        ListArray::<O>::new(
-            self.arrays[0].data_type().clone(),
-            offsets.into(),
-            values,
-            validity.into(),
-        )
+        #[cfg(debug_assertions)]
+        {
+            crate::array::specification::try_check_offsets(&offsets, values.len()).unwrap();
+        }
+
+        // Safety - the invariant of this struct ensures that this is up-held
+        unsafe {
+            ListArray::<O>::new_unchecked(
+                self.arrays[0].data_type().clone(),
+                offsets.into(),
+                values,
+                validity.into(),
+            )
+        }
     }
 }
 
@@ -135,15 +143,7 @@ impl<'a, O: Offset> Growable<'a> for GrowableList<'a, O> {
 }
 
 impl<'a, O: Offset> From<GrowableList<'a, O>> for ListArray<O> {
-    fn from(val: GrowableList<'a, O>) -> Self {
-        let mut values = val.values;
-        let values = values.as_box();
-
-        ListArray::<O>::new(
-            val.arrays[0].data_type().clone(),
-            val.offsets.into(),
-            values,
-            val.validity.into(),
-        )
+    fn from(mut val: GrowableList<'a, O>) -> Self {
+        val.to()
     }
 }
diff --git a/src/array/growable/utf8.rs b/src/array/growable/utf8.rs
index c50c6fd4db..adfdba2cd5 100644
--- a/src/array/growable/utf8.rs
+++ b/src/array/growable/utf8.rs
@@ -55,13 +55,19 @@ impl<'a, O: Offset> GrowableUtf8<'a, O> {
         let offsets = std::mem::take(&mut self.offsets);
         let values = std::mem::take(&mut self.values);
 
+        #[cfg(debug_assertions)]
+        {
+            crate::array::specification::try_check_offsets_and_utf8(&offsets, &values).unwrap();
+        }
+
         unsafe {
-            Utf8Array::<O>::from_data_unchecked(
+            Utf8Array::<O>::try_new_unchecked(
                 self.arrays[0].data_type().clone(),
                 offsets.into(),
                 values.into(),
                 validity.into(),
             )
+            .unwrap()
         }
     }
 }
@@ -99,14 +105,7 @@ impl<'a, O: Offset> Growable<'a> for GrowableUtf8<'a, O> {
 }
 
 impl<'a, O: Offset> From<GrowableUtf8<'a, O>> for Utf8Array<O> {
-    fn from(val: GrowableUtf8<'a, O>) -> Self {
-        unsafe {
-            Utf8Array::<O>::from_data_unchecked(
-                val.arrays[0].data_type().clone(),
-                val.offsets.into(),
-                val.values.into(),
-                val.validity.into(),
-            )
-        }
+    fn from(mut val: GrowableUtf8<'a, O>) -> Self {
+        val.to()
     }
 }
diff --git a/src/array/list/ffi.rs b/src/array/list/ffi.rs
index 717ca06383..c6f1b2985d 100644
--- a/src/array/list/ffi.rs
+++ b/src/array/list/ffi.rs
@@ -4,7 +4,7 @@ use super::super::{ffi::ToFfi, Array, Offset};
 use super::ListArray;
 
 unsafe impl<O: Offset> ToFfi for ListArray<O> {
-    fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
+    fn buffers(&self) -> Vec<Option<*const u8>> {
         vec![
             self.validity.as_ref().map(|x| x.as_ptr()),
             Some(self.offsets.as_ptr().cast::<u8>()),
diff --git a/src/array/map/ffi.rs b/src/array/map/ffi.rs
index 538da4cbc3..bbf3846999 100644
--- a/src/array/map/ffi.rs
+++ b/src/array/map/ffi.rs
@@ -4,7 +4,7 @@ use super::super::{ffi::ToFfi, Array};
 use super::MapArray;
 
 unsafe impl ToFfi for MapArray {
-    fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
+    fn buffers(&self) -> Vec<Option<*const u8>> {
         vec![
             self.validity.as_ref().map(|x| x.as_ptr()),
             Some(self.offsets.as_ptr().cast::<u8>()),
diff --git a/src/array/null.rs b/src/array/null.rs
index 6a00fcb353..041b59ada1 100644
--- a/src/array/null.rs
+++ b/src/array/null.rs
@@ -127,7 +127,7 @@ impl std::fmt::Debug for NullArray {
 }
 
 unsafe impl ToFfi for NullArray {
-    fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
+    fn buffers(&self) -> Vec<Option<*const u8>> {
         // `None` is technically not required by the specification, but older C++ implementations require it, so leaving
         // it here for backward compatibility
         vec![None]
diff --git a/src/array/primitive/ffi.rs b/src/array/primitive/ffi.rs
index f1a7369227..de5d6a7058 100644
--- a/src/array/primitive/ffi.rs
+++ b/src/array/primitive/ffi.rs
@@ -10,7 +10,7 @@ use crate::error::Result;
 use super::PrimitiveArray;
 
 unsafe impl<T: NativeType> ToFfi for PrimitiveArray<T> {
-    fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
+    fn buffers(&self) -> Vec<Option<*const u8>> {
         vec![
             self.validity.as_ref().map(|x| x.as_ptr()),
             Some(self.values.as_ptr().cast::<u8>()),
diff --git a/src/array/struct_/ffi.rs b/src/array/struct_/ffi.rs
index 61b36ca1b8..b29a342d88 100644
--- a/src/array/struct_/ffi.rs
+++ b/src/array/struct_/ffi.rs
@@ -3,7 +3,7 @@ use super::StructArray;
 use crate::{error::Result, ffi};
 
 unsafe impl ToFfi for StructArray {
-    fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
+    fn buffers(&self) -> Vec<Option<*const u8>> {
         vec![self.validity.as_ref().map(|x| x.as_ptr())]
     }
 
diff --git a/src/array/struct_/mod.rs b/src/array/struct_/mod.rs
index e7e99a9c60..2c9383c1ec 100644
--- a/src/array/struct_/mod.rs
+++ b/src/array/struct_/mod.rs
@@ -82,7 +82,7 @@ impl StructArray {
                 if a_len != len {
                     Err(Error::oos(format!(
                         "The children DataTypes of a StructArray must equal the children data types.
-                         However, the values {index} has a length of {a_len}, which is different from values 0, {len}."
+                         However, the values at index {index} have a length of {a_len}, which is different from values at index 0, {len}."
                     )))
                 } else {
                     Ok(())
diff --git a/src/array/union/ffi.rs b/src/array/union/ffi.rs
index 94bcdea42c..fd12f15b23 100644
--- a/src/array/union/ffi.rs
+++ b/src/array/union/ffi.rs
@@ -4,7 +4,7 @@ use super::super::{ffi::ToFfi, Array};
 use super::UnionArray;
 
 unsafe impl ToFfi for UnionArray {
-    fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
+    fn buffers(&self) -> Vec<Option<*const u8>> {
         if let Some(offsets) = &self.offsets {
             vec![
                 Some(self.types.as_ptr().cast::<u8>()),
diff --git a/src/array/union/mod.rs b/src/array/union/mod.rs
index 6d605da064..3bda591568 100644
--- a/src/array/union/mod.rs
+++ b/src/array/union/mod.rs
@@ -1,4 +1,4 @@
-use std::collections::HashMap;
+use ahash::AHashMap;
 
 use crate::{
     bitmap::Bitmap,
@@ -31,7 +31,7 @@ type UnionComponents<'a> = (&'a [Field], Option<&'a [i32]>, UnionMode);
 pub struct UnionArray {
     types: Buffer<i8>,
     // None represents when there is no typeid
-    fields_hash: Option<HashMap<i8, FieldEntry>>,
+    fields_hash: Option<AHashMap<i8, FieldEntry>>,
     fields: Vec<Box<dyn Array>>,
     offsets: Option<Buffer<i32>>,
     data_type: DataType,
diff --git a/src/array/utf8/ffi.rs b/src/array/utf8/ffi.rs
index e8caa53b65..abaed6b28e 100644
--- a/src/array/utf8/ffi.rs
+++ b/src/array/utf8/ffi.rs
@@ -8,7 +8,7 @@ use crate::{
 use super::Utf8Array;
 
 unsafe impl<O: Offset> ToFfi for Utf8Array<O> {
-    fn buffers(&self) -> Vec<Option<std::ptr::NonNull<u8>>> {
+    fn buffers(&self) -> Vec<Option<*const u8>> {
         vec![
             self.validity.as_ref().map(|x| x.as_ptr()),
             Some(self.offsets.as_ptr().cast::<u8>()),
diff --git a/src/bitmap/immutable.rs b/src/bitmap/immutable.rs
index 5384ae5ec1..f4a7cc7df3 100644
--- a/src/bitmap/immutable.rs
+++ b/src/bitmap/immutable.rs
@@ -1,8 +1,8 @@
+use std::{iter::FromIterator, ops::Deref, sync::Arc};
+
 use either::Either;
-use std::iter::FromIterator;
-use std::sync::Arc;
 
-use crate::{buffer::bytes::Bytes, error::Error, trusted_len::TrustedLen};
+use crate::{buffer::Bytes, error::Error, trusted_len::TrustedLen};
 
 use super::{
     chunk_iter_to_vec,
@@ -216,8 +216,8 @@ impl Bitmap {
 
     /// Returns a pointer to the start of this [`Bitmap`] (ignores `offsets`)
     /// This pointer is allocated iff `self.len() > 0`.
-    pub(crate) fn as_ptr(&self) -> std::ptr::NonNull<u8> {
-        self.bytes.ptr()
+    pub(crate) fn as_ptr(&self) -> *const u8 {
+        self.bytes.deref().as_ptr()
     }
 
     /// Returns a pointer to the start of this [`Bitmap`] (ignores `offsets`)
diff --git a/src/buffer/bytes.rs b/src/buffer/bytes.rs
deleted file mode 100644
index 67ad7bed13..0000000000
--- a/src/buffer/bytes.rs
+++ /dev/null
@@ -1,118 +0,0 @@
-use std::mem::ManuallyDrop;
-use std::ops::{Deref, DerefMut};
-use std::panic::RefUnwindSafe;
-use std::ptr::NonNull;
-
-/// Mode of deallocating memory regions
-enum Allocation {
-    /// Native allocation
-    Native,
-    // A foreign allocator and its ref count
-    Foreign(Box<dyn RefUnwindSafe + Send + Sync>),
-}
-
-/// A continuous memory region that may be allocated externally.
-///
-/// In the most common case, this is created from [`Vec`].
-/// However, this region can also be allocated by a foreign allocator.
-pub struct Bytes<T> {
-    /// An implementation using an `enum` of a `Vec` or a foreign pointer is not used
-    /// because `deref` is at least 50% more expensive than the deref of a `Vec`.
-    data: ManuallyDrop<Vec<T>>,
-    /// the region was allocated
-    allocation: Allocation,
-}
-
-impl<T> Bytes<T> {
-    /// Takes ownership of an allocated memory region `[ptr, ptr+len[`,
-    /// # Safety
-    /// This function is safe iff:
-    /// * the region is properly allocated in that a slice can be safely built from it.
-    /// * the region is immutable.
-    /// # Implementation
-    /// This function leaks iff `owner` does not deallocate the region when dropped.
-    #[inline]
-    pub unsafe fn from_owned(
-        ptr: std::ptr::NonNull<T>,
-        len: usize,
-        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
-        // is that we do not expose this region as a `Vec`; we only use `Vec` on it to provide
-        // immutable access to the region (via `Vec::deref` to `&[T]`).
-        // MIRI does not complain, which seems to agree with the line of thought.
-        let data = Vec::from_raw_parts(ptr.as_ptr(), len, len);
-        let data = ManuallyDrop::new(data);
-
-        Self {
-            data,
-            allocation: Allocation::Foreign(owner),
-        }
-    }
-
-    /// The length of the region
-    #[inline]
-    pub fn len(&self) -> usize {
-        self.data.len()
-    }
-
-    /// The pointer to the region
-    #[inline]
-    pub fn ptr(&self) -> NonNull<T> {
-        unsafe { NonNull::new_unchecked(self.data.as_ptr() as *mut T) }
-    }
-
-    /// Returns a `Some` mutable reference of [`Vec<T>`] iff this was initialized
-    /// from a [`Vec<T>`] and `None` otherwise.
-    #[inline]
-    pub fn get_vec(&mut self) -> Option<&mut Vec<T>> {
-        match &self.allocation {
-            Allocation::Foreign(_) => None,
-            Allocation::Native => Some(self.data.deref_mut()),
-        }
-    }
-}
-
-impl<T> Drop for Bytes<T> {
-    #[inline]
-    fn drop(&mut self) {
-        match self.allocation {
-            Allocation::Foreign(_) => {
-                // The ref count of the foreign is reduced by one
-                // we can't deallocate `Vec` here since the region was allocated by
-                // a foreign allocator
-            }
-            Allocation::Native => {
-                let data = std::mem::take(&mut self.data);
-                let _ = ManuallyDrop::into_inner(data);
-            }
-        }
-    }
-}
-
-impl<T> std::ops::Deref for Bytes<T> {
-    type Target = [T];
-
-    #[inline]
-    fn deref(&self) -> &[T] {
-        &self.data
-    }
-}
-
-impl<T: PartialEq> PartialEq for Bytes<T> {
-    #[inline]
-    fn eq(&self, other: &Bytes<T>) -> bool {
-        self.deref() == other.deref()
-    }
-}
-
-impl<T> From<Vec<T>> for Bytes<T> {
-    #[inline]
-    fn from(data: Vec<T>) -> Self {
-        Self {
-            data: ManuallyDrop::new(data),
-            allocation: Allocation::Native,
-        }
-    }
-}
diff --git a/src/buffer/immutable.rs b/src/buffer/immutable.rs
index 8bbe0115d2..1d7e77add6 100644
--- a/src/buffer/immutable.rs
+++ b/src/buffer/immutable.rs
@@ -1,8 +1,8 @@
-use std::{iter::FromIterator, sync::Arc, usize};
+use std::{iter::FromIterator, ops::Deref, sync::Arc, usize};
 
 use crate::types::NativeType;
 
-use super::bytes::Bytes;
+use super::Bytes;
 
 /// [`Buffer`] is a contiguous memory region of plain old data types
 /// that can be shared across thread boundaries.
@@ -33,7 +33,7 @@ use super::bytes::Bytes;
 /// // but cloning forbids getting mut since `slice` and `buffer` now share data
 /// assert_eq!(buffer.get_mut(), None);
 /// ```
-#[derive(Clone, PartialEq)]
+#[derive(Clone)]
 pub struct Buffer<T: NativeType> {
     /// the internal byte buffer.
     data: Arc<Bytes<T>>,
@@ -46,6 +46,13 @@ pub struct Buffer<T: NativeType> {
     length: usize,
 }
 
+impl<T: NativeType> PartialEq for Buffer<T> {
+    #[inline]
+    fn eq(&self, other: &Self) -> bool {
+        self.deref() == other.deref()
+    }
+}
+
 impl<T: NativeType> std::fmt::Debug for Buffer<T> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         std::fmt::Debug::fmt(&**self, f)
@@ -127,8 +134,8 @@ impl<T: NativeType> Buffer<T> {
 
     /// Returns a pointer to the start of this buffer.
     #[inline]
-    pub(crate) fn as_ptr(&self) -> std::ptr::NonNull<T> {
-        self.data.ptr()
+    pub(crate) fn as_ptr(&self) -> *const T {
+        self.data.deref().as_ptr()
     }
 
     /// Returns the offset of this buffer.
diff --git a/src/buffer/mod.rs b/src/buffer/mod.rs
index 0868908e82..10f6248870 100644
--- a/src/buffer/mod.rs
+++ b/src/buffer/mod.rs
@@ -2,6 +2,8 @@
 
 mod immutable;
 
-pub(crate) mod bytes;
+use crate::ffi::InternalArrowArray;
+
+pub(crate) type Bytes<T> = foreign_vec::ForeignVec<Box<InternalArrowArray>, T>;
 
 pub use immutable::Buffer;
diff --git a/src/compute/cast/dictionary_to.rs b/src/compute/cast/dictionary_to.rs
index 01d3325cf9..101669f644 100644
--- a/src/compute/cast/dictionary_to.rs
+++ b/src/compute/cast/dictionary_to.rs
@@ -7,7 +7,7 @@ use crate::{
 };
 
 macro_rules! key_cast {
-    ($keys:expr, $values:expr, $array:expr, $to_keys_type:expr, $to_type:ty) => {{
+    ($keys:expr, $values:expr, $array:expr, $to_keys_type:expr, $to_type:ty, $to_datatype:expr) => {{
         let cast_keys = primitive_to_primitive::<_, $to_type>($keys, $to_keys_type);
 
         // Failure to cast keys (because they don't fit in the
@@ -15,7 +15,11 @@ macro_rules! key_cast {
         if cast_keys.null_count() > $keys.null_count() {
             return Err(Error::Overflow);
         }
-        DictionaryArray::try_new($array.data_type().clone(), $keys.clone(), $values.clone())
+        // Safety: this is safe because given a type `T` that fits in a `usize`, casting it to type `P` either overflows or also fits in a `usize`
+        unsafe {
+             DictionaryArray::try_new_unchecked($to_datatype, cast_keys, $values.clone())
+        }
+            .map(|x| x.boxed())
     }};
 }
 
@@ -88,8 +92,8 @@ where
             Box::new(values.data_type().clone()),
             is_ordered,
         );
-        // some of the values may not fit in `usize` and thus this needs to be checked
-        DictionaryArray::try_new(data_type, casted_keys, values.clone())
+        // Safety: this is safe because given a type `T` that fits in a `usize`, casting it to type `P` either overflows or also fits in a `usize`
+        unsafe { DictionaryArray::try_new_unchecked(data_type, casted_keys, values.clone()) }
     }
 }
 
@@ -134,11 +138,13 @@ pub(super) fn dictionary_cast_dyn<K: DictionaryKey + num_traits::NumCast>(
             let values = cast(values.as_ref(), to_values_type, options)?;
 
             // create the appropriate array type
-            let data_type = (*to_keys_type).into();
+            let to_key_type = (*to_keys_type).into();
+
+            // Safety:
+            // we return an error on overflow so the integers remain within bounds
             match_integer_type!(to_keys_type, |$T| {
-                key_cast!(keys, values, array, &data_type, $T)
+                key_cast!(keys, values, array, &to_key_type, $T, to_type.clone())
             })
-            .map(|x| x.boxed())
         }
         _ => unpack_dictionary::<K>(keys, values.as_ref(), to_type, options),
     }
diff --git a/src/compute/like.rs b/src/compute/like.rs
index 4509617398..cc7517e823 100644
--- a/src/compute/like.rs
+++ b/src/compute/like.rs
@@ -1,6 +1,6 @@
 //! Contains "like" operators such as [`like_utf8`] and [`like_utf8_scalar`].
-use std::collections::HashMap;
 
+use ahash::AHashMap;
 use regex::bytes::Regex as BytesRegex;
 use regex::Regex;
 
@@ -35,7 +35,7 @@ fn a_like_utf8<O: Offset, F: Fn(bool) -> bool>(
 
     let validity = combine_validities(lhs.validity(), rhs.validity());
 
-    let mut map = HashMap::new();
+    let mut map = AHashMap::new();
 
     let values =
         Bitmap::try_from_trusted_len_iter(lhs.iter().zip(rhs.iter()).map(|(lhs, rhs)| {
@@ -179,7 +179,7 @@ fn a_like_binary<O: Offset, F: Fn(bool) -> bool>(
 
     let validity = combine_validities(lhs.validity(), rhs.validity());
 
-    let mut map = HashMap::new();
+    let mut map = AHashMap::new();
 
     let values =
         Bitmap::try_from_trusted_len_iter(lhs.iter().zip(rhs.iter()).map(|(lhs, rhs)| {
diff --git a/src/compute/merge_sort/mod.rs b/src/compute/merge_sort/mod.rs
index 9c48a29dd0..2776d58266 100644
--- a/src/compute/merge_sort/mod.rs
+++ b/src/compute/merge_sort/mod.rs
@@ -56,8 +56,9 @@
 //! To serialize slices, e.g. for checkpointing or transfer via Arrow's IPC, you can store
 //! them as 3 non-null primitive arrays (e.g. `PrimitiveArray<i64>`).
 
+use ahash::AHashMap;
+use std::cmp::Ordering;
 use std::iter::once;
-use std::{cmp::Ordering, collections::HashMap};
 
 use itertools::Itertools;
 
@@ -498,7 +499,7 @@ pub fn build_comparator_impl<'a>(
                 .collect::<Result<Vec<_>>>()?;
             Ok(((lhs_index, rhs_index), multi_column_comparator))
         })
-        .collect::<Result<HashMap<(usize, usize), Vec<(IsValid, IsValid, DynComparator)>>>>()?;
+        .collect::<Result<AHashMap<(usize, usize), Vec<(IsValid, IsValid, DynComparator)>>>>()?;
 
     // prepare a comparison function taking into account _nulls_ and sort options
     let cmp = move |left_index, left_row, right_index, right_row| {
diff --git a/src/compute/regex_match.rs b/src/compute/regex_match.rs
index 001bd0f5d2..41cacb293b 100644
--- a/src/compute/regex_match.rs
+++ b/src/compute/regex_match.rs
@@ -1,7 +1,6 @@
 //! Contains regex matching operators [`regex_match`] and [`regex_match_scalar`].
 
-use std::collections::HashMap;
-
+use ahash::AHashMap;
 use regex::Regex;
 
 use super::utils::combine_validities;
@@ -18,7 +17,7 @@ pub fn regex_match<O: Offset>(values: &Utf8Array<O>, regex: &Utf8Array<O>) -> Re
         ));
     }
 
-    let mut map = HashMap::new();
+    let mut map = AHashMap::new();
     let validity = combine_validities(values.validity(), regex.validity());
 
     let iterator = values.iter().zip(regex.iter()).map(|(haystack, regex)| {
diff --git a/src/ffi/array.rs b/src/ffi/array.rs
index 6bce816255..fd45ce9652 100644
--- a/src/ffi/array.rs
+++ b/src/ffi/array.rs
@@ -5,7 +5,7 @@ use std::sync::Arc;
 use crate::{
     array::*,
     bitmap::{utils::bytes_for, Bitmap},
-    buffer::{bytes::Bytes, Buffer},
+    buffer::{Buffer, Bytes},
     datatypes::{DataType, PhysicalType},
     error::{Error, Result},
     ffi::schema::get_child,
@@ -101,7 +101,7 @@ impl ArrowArray {
         let buffers_ptr = buffers
             .iter()
             .map(|maybe_buffer| match maybe_buffer {
-                Some(b) => b.as_ptr() as *const std::os::raw::c_void,
+                Some(b) => *b as *const std::os::raw::c_void,
                 None => std::ptr::null(),
             })
             .collect::<Box<[_]>>();
@@ -195,7 +195,7 @@ unsafe fn create_buffer<T: NativeType>(
     let len = buffer_len(array, data_type, index)?;
     let offset = buffer_offset(array, data_type, index);
     let bytes = ptr
-        .map(|ptr| Bytes::from_owned(ptr, len, owner))
+        .map(|ptr| Bytes::from_foreign(ptr.as_ptr(), len, owner))
         .ok_or_else(|| Error::OutOfSpec(format!("The buffer at position {} is null", index)))?;
 
     Ok(Buffer::from_bytes(bytes).slice(offset, len - offset))
@@ -226,7 +226,7 @@ unsafe fn create_bitmap(
     let bytes_len = bytes_for(offset + len);
     let ptr = NonNull::new(ptr as *mut u8);
     let bytes = ptr
-        .map(|ptr| Bytes::from_owned(ptr, bytes_len, owner))
+        .map(|ptr| Bytes::from_foreign(ptr.as_ptr(), bytes_len, owner))
         .ok_or_else(|| {
             Error::OutOfSpec(format!(
                 "The buffer {} is a null pointer and cannot be interpreted as a bitmap",
diff --git a/src/io/avro/mod.rs b/src/io/avro/mod.rs
index 852915d900..5addae95e4 100644
--- a/src/io/avro/mod.rs
+++ b/src/io/avro/mod.rs
@@ -48,7 +48,7 @@ macro_rules! avro_decode {
 
 macro_rules! read_header {
     ($reader:ident $($_await:tt)*) => {{
-        let mut items = HashMap::new();
+        let mut items = ahash::AHashMap::new();
 
         loop {
             let len = zigzag_i64($reader)$($_await)*? as usize;
diff --git a/src/io/avro/read/header.rs b/src/io/avro/read/header.rs
index 66bec96cc0..c985ce2508 100644
--- a/src/io/avro/read/header.rs
+++ b/src/io/avro/read/header.rs
@@ -1,5 +1,4 @@
-use std::collections::HashMap;
-
+use ahash::AHashMap;
 use avro_schema::Schema;
 use serde_json;
 
@@ -9,7 +8,7 @@ use super::Compression;
 
 /// Deserializes the Avro header into an Avro [`Schema`] and optional [`Compression`].
 pub(crate) fn deserialize_header(
-    header: HashMap<String, Vec<u8>>,
+    header: AHashMap<String, Vec<u8>>,
 ) -> Result<(Schema, Option<Compression>)> {
     let schema = header
         .get("avro.schema")
diff --git a/src/io/avro/read/util.rs b/src/io/avro/read/util.rs
index 92cbe3db2b..b14870a390 100644
--- a/src/io/avro/read/util.rs
+++ b/src/io/avro/read/util.rs
@@ -1,4 +1,4 @@
-use std::collections::HashMap;
+use ahash::AHashMap;
 use std::io::Read;
 
 use avro_schema::Schema;
@@ -29,7 +29,7 @@ fn _read_binary<R: Read>(reader: &mut R) -> Result<Vec<u8>> {
     Ok(buf)
 }
 
-fn read_header<R: Read>(reader: &mut R) -> Result<HashMap<String, Vec<u8>>> {
+fn read_header<R: Read>(reader: &mut R) -> Result<AHashMap<String, Vec<u8>>> {
     read_header!(reader)
 }
 
diff --git a/src/io/avro/read_async/metadata.rs b/src/io/avro/read_async/metadata.rs
index 4609509376..213428e0f0 100644
--- a/src/io/avro/read_async/metadata.rs
+++ b/src/io/avro/read_async/metadata.rs
@@ -1,6 +1,5 @@
 //! Async Avro
-use std::collections::HashMap;
-
+use ahash::AHashMap;
 use avro_schema::{Record, Schema as AvroSchema};
 use futures::AsyncRead;
 use futures::AsyncReadExt;
@@ -56,6 +55,6 @@ async fn _read_binary<R: AsyncRead + Unpin + Send>(reader: &mut R) -> Result<Vec
 
 async fn read_header<R: AsyncRead + Unpin + Send>(
     reader: &mut R,
-) -> Result<HashMap<String, Vec<u8>>> {
+) -> Result<AHashMap<String, Vec<u8>>> {
     read_header!(reader.await)
 }
diff --git a/src/io/avro/write/header.rs b/src/io/avro/write/header.rs
index 1d25f2f596..370a0532b5 100644
--- a/src/io/avro/write/header.rs
+++ b/src/io/avro/write/header.rs
@@ -1,5 +1,4 @@
-use std::collections::HashMap;
-
+use ahash::AHashMap;
 use avro_schema::Schema;
 use serde_json;
 
@@ -11,10 +10,10 @@ use super::Compression;
 pub(crate) fn serialize_header(
     schema: &Schema,
     compression: Option<Compression>,
-) -> Result<HashMap<String, Vec<u8>>> {
+) -> Result<AHashMap<String, Vec<u8>>> {
     let schema = serde_json::to_string(schema).map_err(|e| Error::ExternalFormat(e.to_string()))?;
 
-    let mut header = HashMap::<String, Vec<u8>>::default();
+    let mut header = AHashMap::<String, Vec<u8>>::default();
 
     header.insert("avro.schema".to_string(), schema.into_bytes());
     if let Some(compression) = compression {
diff --git a/src/io/csv/read/infer_schema.rs b/src/io/csv/read/infer_schema.rs
index 83462bc1e8..4aa3eaf53e 100644
--- a/src/io/csv/read/infer_schema.rs
+++ b/src/io/csv/read/infer_schema.rs
@@ -1,7 +1,5 @@
-use std::{
-    collections::HashSet,
-    io::{Read, Seek},
-};
+use ahash::AHashSet;
+use std::io::{Read, Seek};
 
 use crate::datatypes::{DataType, Field};
 use crate::error::Result;
@@ -34,7 +32,7 @@ pub fn infer_schema<R: Read + Seek, F: Fn(&[u8]) -> DataType>(
 
     let header_length = headers.len();
     // keep track of inferred field types
-    let mut column_types: Vec<HashSet<DataType>> = vec![HashSet::new(); header_length];
+    let mut column_types: Vec<AHashSet<DataType>> = vec![AHashSet::new(); header_length];
 
     let mut records_count = 0;
 
diff --git a/src/io/csv/read_async/infer_schema.rs b/src/io/csv/read_async/infer_schema.rs
index b22f2a613d..2a2a806159 100644
--- a/src/io/csv/read_async/infer_schema.rs
+++ b/src/io/csv/read_async/infer_schema.rs
@@ -1,6 +1,5 @@
-use std::collections::HashSet;
-
 use super::{AsyncReader, ByteRecord};
+use ahash::AHashSet;
 
 use crate::datatypes::{DataType, Field};
 use crate::error::Result;
@@ -41,7 +40,7 @@ where
 
     let header_length = headers.len();
     // keep track of inferred field types
-    let mut column_types: Vec<HashSet<DataType>> = vec![HashSet::new(); header_length];
+    let mut column_types: Vec<AHashSet<DataType>> = vec![AHashSet::new(); header_length];
 
     let mut records_count = 0;
 
diff --git a/src/io/csv/utils.rs b/src/io/csv/utils.rs
index 266801d940..436c7d192f 100644
--- a/src/io/csv/utils.rs
+++ b/src/io/csv/utils.rs
@@ -1,6 +1,5 @@
-use std::collections::HashSet;
-
 use crate::datatypes::{DataType, Field, TimeUnit};
+use ahash::AHashSet;
 
 pub(super) const RFC3339: &str = "%Y-%m-%dT%H:%M:%S%.f%:z";
 
@@ -78,7 +77,7 @@ pub fn infer(bytes: &[u8]) -> DataType {
     }
 }
 
-fn merge_fields(field_name: &str, possibilities: &mut HashSet<DataType>) -> Field {
+fn merge_fields(field_name: &str, possibilities: &mut AHashSet<DataType>) -> Field {
     // determine data type based on possible types
     // if there are incompatible types, use DataType::Utf8
     let data_type = match possibilities.len() {
@@ -101,7 +100,7 @@ fn merge_fields(field_name: &str, possibilities: &mut HashSet<DataType>) -> Fiel
 
 pub(crate) fn merge_schema(
     headers: &[String],
-    column_types: &mut [HashSet<DataType>],
+    column_types: &mut [AHashSet<DataType>],
 ) -> Vec<Field> {
     headers
         .iter()
diff --git a/src/io/ipc/read/common.rs b/src/io/ipc/read/common.rs
index fe445f2079..5638a5289e 100644
--- a/src/io/ipc/read/common.rs
+++ b/src/io/ipc/read/common.rs
@@ -1,4 +1,5 @@
-use std::collections::{HashMap, HashSet, VecDeque};
+use ahash::AHashMap;
+use std::collections::VecDeque;
 use std::io::{Read, Seek};
 
 use arrow_format;
@@ -322,40 +323,47 @@ mod tests {
 pub fn prepare_projection(
     fields: &[Field],
     mut projection: Vec<usize>,
-) -> (Vec<usize>, HashMap<usize, usize>, Vec<Field>) {
-    assert_eq!(
-        projection.iter().collect::<HashSet<_>>().len(),
-        projection.len(),
-        "The projection on IPC must not contain duplicates"
-    );
-
+) -> (Vec<usize>, AHashMap<usize, usize>, Vec<Field>) {
     let fields = projection.iter().map(|x| fields[*x].clone()).collect();
 
     // todo: find way to do this more efficiently
     let mut indices = (0..projection.len()).collect::<Vec<_>>();
     indices.sort_unstable_by_key(|&i| &projection[i]);
     let map = indices.iter().copied().enumerate().fold(
-        HashMap::default(),
+        AHashMap::default(),
         |mut acc, (index, new_index)| {
-            if !acc.contains_key(&new_index) {
-                acc.insert(index, new_index);
-            };
+            acc.insert(index, new_index);
             acc
         },
     );
     projection.sort_unstable();
 
+    // check unique
+    if !projection.is_empty() {
+        let mut previous = projection[0];
+
+        for &i in &projection[1..] {
+            assert!(
+                previous < i,
+                "The projection on IPC must not contain duplicates"
+            );
+            previous = i;
+        }
+    }
+
     (projection, map, fields)
 }
 
 pub fn apply_projection(
     chunk: Chunk<Box<dyn Array>>,
-    map: &HashMap<usize, usize>,
+    map: &AHashMap<usize, usize>,
 ) -> Chunk<Box<dyn Array>> {
     // re-order according to projection
-    let mut arrays = chunk.into_arrays();
-    map.iter().for_each(|(old, new)| {
-        arrays.swap(*old, *new);
-    });
-    Chunk::new(arrays)
+    let arrays = chunk.into_arrays();
+    let mut new_arrays = arrays.clone();
+
+    map.iter()
+        .for_each(|(old, new)| new_arrays[*new] = arrays[*old].clone());
+
+    Chunk::new(new_arrays)
 }
diff --git a/src/io/ipc/read/file_async.rs b/src/io/ipc/read/file_async.rs
index 0d80b80ac6..4b36a44026 100644
--- a/src/io/ipc/read/file_async.rs
+++ b/src/io/ipc/read/file_async.rs
@@ -1,5 +1,5 @@
 //! Async reader for Arrow IPC files
-use std::collections::HashMap;
+use ahash::AHashMap;
 use std::io::SeekFrom;
 
 use arrow_format::ipc::{planus::ReadAsRoot, Block, MessageHeaderRef};
@@ -73,7 +73,7 @@ impl<'a> FileStream<'a> {
         mut reader: R,
         mut dictionaries: Option<Dictionaries>,
         metadata: FileMetadata,
-        projection: Option<(Vec<usize>, HashMap<usize, usize>)>,
+        projection: Option<(Vec<usize>, AHashMap<usize, usize>)>,
         limit: Option<usize>,
     ) -> BoxStream<'a, Result<Chunk<Box<dyn Array>>>>
     where
diff --git a/src/io/ipc/read/mod.rs b/src/io/ipc/read/mod.rs
index 5ffe6426e2..f623e0aef5 100644
--- a/src/io/ipc/read/mod.rs
+++ b/src/io/ipc/read/mod.rs
@@ -4,7 +4,7 @@
 //! which provides arbitrary access to any of its messages, and the
 //! [`StreamReader`](stream::StreamReader), which only supports reading
 //! data in the order it was written in.
-use std::collections::HashMap;
+use ahash::AHashMap;
 
 use crate::array::Array;
 
@@ -35,7 +35,7 @@ pub use schema::deserialize_schema;
 pub use stream::{read_stream_metadata, StreamMetadata, StreamReader, StreamState};
 
 /// how dictionaries are tracked in this crate
-pub type Dictionaries = HashMap<i64, Box<dyn Array>>;
+pub type Dictionaries = AHashMap<i64, Box<dyn Array>>;
 
 pub(crate) type Node<'a> = arrow_format::ipc::FieldNodeRef<'a>;
 pub(crate) type IpcBuffer<'a> = arrow_format::ipc::BufferRef<'a>;
diff --git a/src/io/ipc/read/reader.rs b/src/io/ipc/read/reader.rs
index b18ca4cead..ab72adec96 100644
--- a/src/io/ipc/read/reader.rs
+++ b/src/io/ipc/read/reader.rs
@@ -1,4 +1,4 @@
-use std::collections::HashMap;
+use ahash::AHashMap;
 use std::convert::TryInto;
 use std::io::{Read, Seek, SeekFrom};
 
@@ -119,7 +119,7 @@ pub fn read_file_dictionaries<R: Read + Seek>(
     let blocks = if let Some(blocks) = metadata.dictionaries.as_deref() {
         blocks
     } else {
-        return Ok(HashMap::new());
+        return Ok(AHashMap::new());
     };
     // use a temporary smaller scratch for the messages
     let mut message_scratch = Default::default();
@@ -326,7 +326,7 @@ pub struct FileReader<R: Read + Seek> {
     // the dictionaries are going to be read
     dictionaries: Option<Dictionaries>,
     current_block: usize,
-    projection: Option<(Vec<usize>, HashMap<usize, usize>, Schema)>,
+    projection: Option<(Vec<usize>, AHashMap<usize, usize>, Schema)>,
     remaining: usize,
     data_scratch: Vec<u8>,
     message_scratch: Vec<u8>,
diff --git a/src/io/ipc/read/stream.rs b/src/io/ipc/read/stream.rs
index b3f3a986d6..cea6e08825 100644
--- a/src/io/ipc/read/stream.rs
+++ b/src/io/ipc/read/stream.rs
@@ -1,4 +1,4 @@
-use std::collections::HashMap;
+use ahash::AHashMap;
 use std::io::Read;
 
 use arrow_format;
@@ -96,7 +96,7 @@ fn read_next<R: Read>(
     dictionaries: &mut Dictionaries,
     message_buffer: &mut Vec<u8>,
     data_buffer: &mut Vec<u8>,
-    projection: &Option<(Vec<usize>, HashMap<usize, usize>, Schema)>,
+    projection: &Option<(Vec<usize>, AHashMap<usize, usize>, Schema)>,
     scratch: &mut Vec<u8>,
 ) -> Result<Option<StreamState>> {
     // determine metadata length
@@ -241,7 +241,7 @@ pub struct StreamReader<R: Read> {
     finished: bool,
     data_buffer: Vec<u8>,
     message_buffer: Vec<u8>,
-    projection: Option<(Vec<usize>, HashMap<usize, usize>, Schema)>,
+    projection: Option<(Vec<usize>, AHashMap<usize, usize>, Schema)>,
     scratch: Vec<u8>,
 }
 
diff --git a/src/io/json_integration/read/array.rs b/src/io/json_integration/read/array.rs
index 2d7d0ec39a..0c44cc2432 100644
--- a/src/io/json_integration/read/array.rs
+++ b/src/io/json_integration/read/array.rs
@@ -1,5 +1,4 @@
-use std::collections::HashMap;
-
+use ahash::AHashMap;
 use num_traits::NumCast;
 use serde_json::Value;
 
@@ -192,7 +191,7 @@ fn to_list<O: Offset>(
     json_col: &ArrowJsonColumn,
     data_type: DataType,
     field: &IpcField,
-    dictionaries: &HashMap<i64, ArrowJsonDictionaryBatch>,
+    dictionaries: &AHashMap<i64, ArrowJsonDictionaryBatch>,
 ) -> Result<Box<dyn Array>> {
     let validity = to_validity(&json_col.validity);
 
@@ -214,7 +213,7 @@ fn to_map(
     json_col: &ArrowJsonColumn,
     data_type: DataType,
     field: &IpcField,
-    dictionaries: &HashMap<i64, ArrowJsonDictionaryBatch>,
+    dictionaries: &AHashMap<i64, ArrowJsonDictionaryBatch>,
 ) -> Result<Box<dyn Array>> {
     let validity = to_validity(&json_col.validity);
 
@@ -234,7 +233,7 @@ fn to_dictionary<K: DictionaryKey + NumCast>(
     data_type: DataType,
     field: &IpcField,
     json_col: &ArrowJsonColumn,
-    dictionaries: &HashMap<i64, ArrowJsonDictionaryBatch>,
+    dictionaries: &AHashMap<i64, ArrowJsonDictionaryBatch>,
 ) -> Result<Box<dyn Array>> {
     // find dictionary
     let dict_id = field.dictionary_id.unwrap();
@@ -260,7 +259,7 @@ pub fn to_array(
     data_type: DataType,
     field: &IpcField,
     json_col: &ArrowJsonColumn,
-    dictionaries: &HashMap<i64, ArrowJsonDictionaryBatch>,
+    dictionaries: &AHashMap<i64, ArrowJsonDictionaryBatch>,
 ) -> Result<Box<dyn Array>> {
     use PhysicalType::*;
     match data_type.to_physical_type() {
@@ -409,7 +408,7 @@ pub fn deserialize_chunk(
     schema: &Schema,
     ipc_fields: &[IpcField],
     json_batch: &ArrowJsonBatch,
-    json_dictionaries: &HashMap<i64, ArrowJsonDictionaryBatch>,
+    json_dictionaries: &AHashMap<i64, ArrowJsonDictionaryBatch>,
 ) -> Result<Chunk<Box<dyn Array>>> {
     let arrays = schema
         .fields
diff --git a/src/io/parquet/read/deserialize/nested_utils.rs b/src/io/parquet/read/deserialize/nested_utils.rs
index e369815d5c..5fafdceee2 100644
--- a/src/io/parquet/read/deserialize/nested_utils.rs
+++ b/src/io/parquet/read/deserialize/nested_utils.rs
@@ -411,9 +411,10 @@ fn extend_offsets2<'a, D: NestedDecoder<'a>>(
         cum_sum[i + 1] = cum_sum[i] + delta;
     }
 
-    let mut is_required = vec![false; nested.len()];
-    for (depth, nest) in nested.iter().enumerate().take(nested.len() - 1) {
-        is_required[depth + 1] = nest.is_required() && nest.is_nullable()
+    let mut cum_rep = vec![0u32; nested.len() + 1];
+    for (i, nest) in nested.iter().enumerate() {
+        let delta = nest.is_repeated() as u32;
+        cum_rep[i + 1] = cum_rep[i] + delta;
     }
 
     let max_depth = nested.len() - 1;
@@ -424,15 +425,21 @@ fn extend_offsets2<'a, D: NestedDecoder<'a>>(
             rows += 1;
         }
 
-        for (depth, (nest, &is_required)) in nested.iter_mut().zip(is_required.iter()).enumerate() {
-            let right_level = depth as u32 >= rep && def >= cum_sum[depth];
+        let mut is_required = false;
+        for (depth, nest) in nested.iter_mut().enumerate() {
+            let right_level = rep <= cum_rep[depth] && def >= cum_sum[depth];
             if is_required || right_level {
-                let is_valid = nest.is_nullable() && def != cum_sum[depth];
+                let is_valid = nest.is_nullable() && def > cum_sum[depth];
                 let length = values_count[depth];
                 nest.push(length, is_valid);
                 if depth > 0 {
                     values_count[depth - 1] = nest.len() as i64;
                 };
+                if nest.is_required() && !is_valid {
+                    is_required = true;
+                } else {
+                    is_required = false
+                };
 
                 if depth == max_depth {
                     // the leaf / primitive
diff --git a/src/io/parquet/write/sink.rs b/src/io/parquet/write/sink.rs
index 160ac81d7f..9ba2d4aea9 100644
--- a/src/io/parquet/write/sink.rs
+++ b/src/io/parquet/write/sink.rs
@@ -1,4 +1,5 @@
-use std::{collections::HashMap, pin::Pin, task::Poll};
+use ahash::AHashMap;
+use std::{pin::Pin, task::Poll};
 
 use futures::{future::BoxFuture, AsyncWrite, FutureExt, Sink, TryFutureExt};
 use parquet2::metadata::KeyValue;
@@ -62,7 +63,7 @@ pub struct FileSink<'a, W: AsyncWrite + Send + Unpin> {
     schema: Schema,
     parquet_schema: SchemaDescriptor,
     /// Key-value metadata that will be written to the file on close.
-    pub metadata: HashMap<String, Option<String>>,
+    pub metadata: AHashMap<String, Option<String>>,
 }
 
 impl<'a, W> FileSink<'a, W>
@@ -105,7 +106,7 @@ where
             schema,
             encodings,
             parquet_schema,
-            metadata: HashMap::default(),
+            metadata: AHashMap::default(),
         })
     }
 
diff --git a/src/lib.rs b/src/lib.rs
index 03963841b5..01fe444d5c 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -37,3 +37,6 @@ mod docs;
 
 // re-exported because we return `Either` in our public API
 pub use either::Either;
+
+// re-exported to construct dictionaries
+pub use ahash::AHashMap;


Test patch:
diff --git a/arrow-parquet-integration-testing/src/main.rs b/arrow-parquet-integration-testing/src/main.rs
index f62fba5e09..d79da94318 100644
--- a/arrow-parquet-integration-testing/src/main.rs
+++ b/arrow-parquet-integration-testing/src/main.rs
@@ -1,9 +1,10 @@
 use std::fs::File;
-use std::{collections::HashMap, io::Read};
+use std::{io::Read};
 
 use arrow2::array::Array;
 use arrow2::io::ipc::IpcField;
 use arrow2::{
+    AHashMap,
     chunk::Chunk,
     datatypes::{DataType, Schema},
     error::Result,
@@ -40,7 +41,7 @@ pub fn read_gzip_json(
     let (schema, ipc_fields) = read::deserialize_schema(&schema)?;
 
     // read dictionaries
-    let mut dictionaries = HashMap::new();
+    let mut dictionaries = AHashMap::new();
     if let Some(dicts) = arrow_json.dictionaries {
         for json_dict in dicts {
             // TODO: convert to a concrete Arrow type
diff --git a/integration-testing/src/lib.rs b/integration-testing/src/lib.rs
index 0a3b778e38..e36ad7eef0 100644
--- a/integration-testing/src/lib.rs
+++ b/integration-testing/src/lib.rs
@@ -22,11 +22,11 @@ use arrow2::io::ipc::IpcField;
 use serde_json::Value;
 
 use arrow2::chunk::Chunk;
+use arrow2::AHashMap;
 use arrow2::datatypes::*;
 use arrow2::error::Result;
 use arrow2::io::json_integration::{read, ArrowJsonBatch, ArrowJsonDictionaryBatch};
 
-use std::collections::HashMap;
 use std::fs::File;
 use std::io::BufReader;
 
@@ -43,7 +43,7 @@ pub struct ArrowFile {
     pub fields: Vec<IpcField>,
     // we can evolve this into a concrete Arrow type
     // this is temporarily not being read from
-    pub _dictionaries: HashMap<i64, ArrowJsonDictionaryBatch>,
+    pub _dictionaries: AHashMap<i64, ArrowJsonDictionaryBatch>,
     pub chunks: Vec<Chunk<Box<dyn Array>>>,
 }
 
@@ -54,7 +54,7 @@ pub fn read_json_file(json_name: &str) -> Result<ArrowFile> {
 
     let (schema, fields) = read::deserialize_schema(&arrow_json["schema"])?;
     // read dictionaries
-    let mut dictionaries = HashMap::new();
+    let mut dictionaries = AHashMap::new();
     if let Some(dicts) = arrow_json.get("dictionaries") {
         for d in dicts
             .as_array()
diff --git a/tests/it/compute/cast.rs b/tests/it/compute/cast.rs
index 79ce42c990..9e9888926d 100644
--- a/tests/it/compute/cast.rs
+++ b/tests/it/compute/cast.rs
@@ -805,3 +805,27 @@ fn utf8_to_date64() {
 
     assert_eq!(&expected, c);
 }
+
+#[test]
+fn dict_keys() {
+    let mut array = MutableDictionaryArray::<u8, MutableUtf8Array<i32>>::new();
+    array
+        .try_extend([Some("one"), None, Some("three"), Some("one")])
+        .unwrap();
+    let array: DictionaryArray<u8> = array.into();
+
+    let result = cast(
+        &array,
+        &DataType::Dictionary(IntegerType::Int64, Box::new(DataType::Utf8), false),
+        CastOptions::default(),
+    )
+    .expect("cast failed");
+
+    let mut expected = MutableDictionaryArray::<i64, MutableUtf8Array<i32>>::new();
+    expected
+        .try_extend([Some("one"), None, Some("three"), Some("one")])
+        .unwrap();
+    let expected: DictionaryArray<i64> = expected.into();
+
+    assert_eq!(expected, result.as_ref());
+}
diff --git a/tests/it/io/ipc/common.rs b/tests/it/io/ipc/common.rs
index deaf44ec6a..6254c8f5bb 100644
--- a/tests/it/io/ipc/common.rs
+++ b/tests/it/io/ipc/common.rs
@@ -1,4 +1,5 @@
-use std::{collections::HashMap, fs::File, io::Read};
+use ahash::AHashMap;
+use std::{fs::File, io::Read};
 
 use arrow2::{
     array::Array, chunk::Chunk, datatypes::Schema, error::Result,
@@ -29,7 +30,7 @@ pub fn read_gzip_json(version: &str, file_name: &str) -> Result<IpcRead> {
     let (schema, ipc_fields) = read::deserialize_schema(&schema)?;
 
     // read dictionaries
-    let mut dictionaries = HashMap::new();
+    let mut dictionaries = AHashMap::new();
     if let Some(dicts) = arrow_json.dictionaries {
         for json_dict in dicts {
             // TODO: convert to a concrete Arrow type
diff --git a/tests/it/io/parquet/mod.rs b/tests/it/io/parquet/mod.rs
index df138d1686..6710bf5769 100644
--- a/tests/it/io/parquet/mod.rs
+++ b/tests/it/io/parquet/mod.rs
@@ -70,6 +70,21 @@ pub fn pyarrow_nested_edge(column: &str) -> Box<dyn Array> {
             let array: ListArray<i32> = a.into();
             Box::new(array)
         }
+        "struct_list_nullable" => {
+            // [["a", "b", None, "c"]]
+            let a = ListArray::<i32>::new(
+                DataType::List(Box::new(Field::new("item", DataType::Utf8, true))),
+                vec![0, 4].into(),
+                Utf8Array::<i32>::from([Some("a"), Some("b"), None, Some("c")]).boxed(),
+                None,
+            );
+            StructArray::new(
+                DataType::Struct(vec![Field::new("f1", a.data_type().clone(), true)]),
+                vec![a.boxed()],
+                None,
+            )
+            .boxed()
+        }
         _ => todo!(),
     }
 }
@@ -675,6 +690,17 @@ pub fn pyarrow_nested_edge_statistics(column: &str) -> Statistics {
         )
     };
 
+    let new_struct = |arrays: Vec<Box<dyn Array>>, names: Vec<String>| {
+        let fields = names
+            .into_iter()
+            .zip(arrays.iter())
+            .map(|(n, a)| Field::new(n, a.data_type().clone(), true))
+            .collect();
+        StructArray::new(DataType::Struct(fields), arrays, None)
+    };
+
+    let names = vec!["f1".to_string()];
+
     match column {
         "simple" => Statistics {
             distinct_count: Count::List(new_list(UInt64Array::from([None]).boxed())),
@@ -688,6 +714,24 @@ pub fn pyarrow_nested_edge_statistics(column: &str) -> Statistics {
             min_value: new_list(Box::new(Int64Array::from([None]))).boxed(),
             max_value: new_list(Box::new(Int64Array::from([None]))).boxed(),
         },
+        "struct_list_nullable" => Statistics {
+            distinct_count: Count::Struct(new_struct(
+                vec![new_list(Box::new(UInt64Array::from([None]))).boxed()],
+                names.clone(),
+            )),
+            null_count: Count::Struct(new_struct(
+                vec![new_list(Box::new(UInt64Array::from([Some(1)]))).boxed()],
+                names.clone(),
+            )),
+            min_value: Box::new(new_struct(
+                vec![new_list(Box::new(Utf8Array::<i32>::from_slice(["a"]))).boxed()],
+                names.clone(),
+            )),
+            max_value: Box::new(new_struct(
+                vec![new_list(Box::new(Utf8Array::<i32>::from_slice(["c"]))).boxed()],
+                names,
+            )),
+        },
         _ => unreachable!(),
     }
 }
@@ -721,6 +765,8 @@ pub fn pyarrow_struct(column: &str) -> Box<dyn Array> {
     ];
     let string = Utf8Array::<i32>::from(string).boxed();
 
+    let mask = [true, true, false, true, true, true, true, true, true, true];
+
     let fields = vec![
         Field::new("f1", DataType::Utf8, true),
         Field::new("f2", DataType::Boolean, true),
@@ -729,12 +775,7 @@ pub fn pyarrow_struct(column: &str) -> Box<dyn Array> {
         "struct" => StructArray::new(DataType::Struct(fields), vec![string, boolean], None).boxed(),
         "struct_nullable" => {
             let values = vec![string, boolean];
-            StructArray::new(
-                DataType::Struct(fields),
-                values,
-                Some([true, true, false, true, true, true, true, true, true, true].into()),
-            )
-            .boxed()
+            StructArray::new(DataType::Struct(fields), values, Some(mask.into())).boxed()
         }
         "struct_struct" => {
             let struct_ = pyarrow_struct("struct");
@@ -747,6 +788,17 @@ pub fn pyarrow_struct(column: &str) -> Box<dyn Array> {
                 None,
             ))
         }
+        "struct_struct_nullable" => {
+            let struct_ = pyarrow_struct("struct");
+            Box::new(StructArray::new(
+                DataType::Struct(vec![
+                    Field::new("f1", DataType::Struct(fields), true),
+                    Field::new("f2", DataType::Boolean, true),
+                ]),
+                vec![struct_, boolean],
+                Some(mask.into()),
+            ))
+        }
         _ => todo!(),
     }
 }
@@ -854,6 +906,66 @@ pub fn pyarrow_struct_statistics(column: &str) -> Statistics {
             )
             .boxed(),
         },
+        "struct_struct_nullable" => Statistics {
+            distinct_count: Count::Struct(new_struct(
+                vec![
+                    new_struct(
+                        vec![
+                            Box::new(UInt64Array::from([None])),
+                            Box::new(UInt64Array::from([None])),
+                        ],
+                        names.clone(),
+                    )
+                    .boxed(),
+                    UInt64Array::from([None]).boxed(),
+                ],
+                names.clone(),
+            )),
+            null_count: Count::Struct(new_struct(
+                vec![
+                    new_struct(
+                        vec![
+                            Box::new(UInt64Array::from([Some(5)])),
+                            Box::new(UInt64Array::from([Some(5)])),
+                        ],
+                        names.clone(),
+                    )
+                    .boxed(),
+                    UInt64Array::from([Some(5)]).boxed(),
+                ],
+                names.clone(),
+            )),
+            min_value: new_struct(
+                vec![
+                    new_struct(
+                        vec![
+                            Utf8Array::<i32>::from_slice([""]).boxed(),
+                            BooleanArray::from_slice([false]).boxed(),
+                        ],
+                        names.clone(),
+                    )
+                    .boxed(),
+                    BooleanArray::from_slice([false]).boxed(),
+                ],
+                names.clone(),
+            )
+            .boxed(),
+            max_value: new_struct(
+                vec![
+                    new_struct(
+                        vec![
+                            Utf8Array::<i32>::from_slice(["def"]).boxed(),
+                            BooleanArray::from_slice([true]).boxed(),
+                        ],
+                        names.clone(),
+                    )
+                    .boxed(),
+                    BooleanArray::from_slice([true]).boxed(),
+                ],
+                names,
+            )
+            .boxed(),
+        },
         _ => todo!(),
     }
 }
diff --git a/tests/it/io/parquet/read.rs b/tests/it/io/parquet/read.rs
index 675ec95c6b..45c5b714d6 100644
--- a/tests/it/io/parquet/read.rs
+++ b/tests/it/io/parquet/read.rs
@@ -462,6 +462,11 @@ fn v1_struct_optional_optional() -> Result<()> {
     test_pyarrow_integration("struct_nullable", 1, "struct", false, false, None)
 }
 
+#[test]
+fn v1_struct_struct_optional() -> Result<()> {
+    test_pyarrow_integration("struct_struct_nullable", 1, "struct", false, false, None)
+}
+
 #[test]
 fn v1_nested_edge_1() -> Result<()> {
     test_pyarrow_integration("simple", 1, "nested_edge", false, false, None)
@@ -472,6 +477,11 @@ fn v1_nested_edge_2() -> Result<()> {
     test_pyarrow_integration("null", 1, "nested_edge", false, false, None)
 }
 
+#[test]
+fn v1_nested_edge_3() -> Result<()> {
+    test_pyarrow_integration("struct_list_nullable", 1, "nested_edge", false, false, None)
+}
+
 #[test]
 fn v1_map() -> Result<()> {
     test_pyarrow_integration("map", 1, "map", false, true, None)
diff --git a/tests/it/io/parquet/write_async.rs b/tests/it/io/parquet/write_async.rs
index 89d76efea9..5644caf67f 100644
--- a/tests/it/io/parquet/write_async.rs
+++ b/tests/it/io/parquet/write_async.rs
@@ -1,5 +1,4 @@
-use std::collections::HashMap;
-
+use ahash::AHashMap;
 use arrow2::{
     array::{Float32Array, Int32Array},
     chunk::Chunk,
@@ -46,7 +45,7 @@ async fn test_parquet_async_roundtrip() {
 
     buffer.set_position(0);
     let metadata = read_metadata_async(&mut buffer).await.unwrap();
-    let kv = HashMap::<String, Option<String>>::from_iter(
+    let kv = AHashMap::<String, Option<String>>::from_iter(
         metadata
             .key_value_metadata()
             .to_owned()