From 4cb8105095fa0757e2b0f44f7f44038bb240f55b Mon Sep 17 00:00:00 2001
From: aykut-bozkurt <51649454+aykut-bozkurt@users.noreply.github.com>
Date: Wed, 11 Sep 2024 15:09:32 +0300
Subject: [PATCH] refactor pg to parquet (#21)

---
 .vscode/settings.json                         |   3 +-
 Cargo.toml                                    |   1 +
 src/arrow_parquet/arrow_utils.rs              |  29 +-
 src/arrow_parquet/parquet_writer.rs           |  57 +-
 src/arrow_parquet/pg_to_arrow.rs              | 498 +++++-------------
 src/arrow_parquet/pg_to_arrow/bool.rs         |  20 +-
 src/arrow_parquet/pg_to_arrow/bytea.rs        |  27 +-
 src/arrow_parquet/pg_to_arrow/char.rs         |  38 +-
 src/arrow_parquet/pg_to_arrow/date.rs         |  38 +-
 .../pg_to_arrow/fallback_to_text.rs           |  38 +-
 src/arrow_parquet/pg_to_arrow/float4.rs       |  20 +-
 src/arrow_parquet/pg_to_arrow/float8.rs       |  21 +-
 src/arrow_parquet/pg_to_arrow/geometry.rs     |  26 +-
 src/arrow_parquet/pg_to_arrow/int2.rs         |  21 +-
 src/arrow_parquet/pg_to_arrow/int4.rs         |  21 +-
 src/arrow_parquet/pg_to_arrow/int8.rs         |  21 +-
 src/arrow_parquet/pg_to_arrow/map.rs          | 103 ++--
 src/arrow_parquet/pg_to_arrow/numeric.rs      |  34 +-
 src/arrow_parquet/pg_to_arrow/oid.rs          |  37 +-
 src/arrow_parquet/pg_to_arrow/record.rs       |  98 ++--
 src/arrow_parquet/pg_to_arrow/text.rs         |  21 +-
 src/arrow_parquet/pg_to_arrow/time.rs         |  38 +-
 src/arrow_parquet/pg_to_arrow/timestamp.rs    |  38 +-
 src/arrow_parquet/pg_to_arrow/timestamptz.rs  |  39 +-
 src/arrow_parquet/pg_to_arrow/timetz.rs       |  38 +-
 src/lib.rs                                    | 158 ++++--
 src/type_compat/pg_arrow_type_conversions.rs  |  32 +-
 27 files changed, 654 insertions(+), 861 deletions(-)

diff --git a/.vscode/settings.json b/.vscode/settings.json
index 6285f13..be4b716 100644
--- a/.vscode/settings.json
+++ b/.vscode/settings.json
@@ -2,5 +2,6 @@
     "editor.formatOnSave": true,
     "editor.defaultFormatter": "rust-lang.rust-analyzer",
     "rust-analyzer.check.command": "clippy",
-    "rust-analyzer.checkOnSave": true
+    "rust-analyzer.checkOnSave": true,
+    "editor.inlayHints.enabled": "offUnlessPressed",
 }
\ No newline at end of file
diff --git a/Cargo.toml b/Cargo.toml
index ba6f061..4976dcb 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -17,6 +17,7 @@ pg_test = []
 
 [dependencies]
 arrow = {version = "53", default-features = false}
+arrow-select = {version = "53", default-features = false}
 arrow-schema = {version = "53", default-features = false}
 futures = "0.3"
 object_store = {version = "0.11", default-features = false, features = ["aws"]}
diff --git a/src/arrow_parquet/arrow_utils.rs b/src/arrow_parquet/arrow_utils.rs
index 0ef0167..243c2bb 100644
--- a/src/arrow_parquet/arrow_utils.rs
+++ b/src/arrow_parquet/arrow_utils.rs
@@ -38,27 +38,22 @@ pub(crate) fn arrow_map_offsets(maps: &Vec<Option<CrunchyMap>>) -> (OffsetBuffer
 }
 
 pub(crate) fn arrow_array_offsets<T>(
-    arrays: &Vec<Option<Vec<Option<T>>>>,
+    pg_array: &Option<pgrx::Array<T>>,
 ) -> (OffsetBuffer<i32>, NullBuffer) {
-    pgrx::pg_sys::check_for_interrupts!();
-
     let mut nulls = vec![];
     let mut offsets = vec![0];
-    let mut current_offset = 0;
-    for array in arrays {
-        if let Some(array) = array {
-            let len = array.len() as i32;
-            current_offset += len;
-            offsets.push(current_offset);
-            nulls.push(true);
-        } else {
-            offsets.push(current_offset);
-            nulls.push(false);
-        }
-    }
 
-    let offsets = OffsetBuffer::new(ScalarBuffer::from(offsets));
-    let nulls = NullBuffer::from(nulls);
+    if let Some(pg_array) = pg_array {
+        let len = pg_array.len() as i32;
+        offsets.push(len);
+        nulls.push(true);
+    } else {
+        offsets.push(0);
+        nulls.push(false);
+    };
+
+    let offsets = arrow::buffer::OffsetBuffer::new(arrow::buffer::ScalarBuffer::from(offsets));
+    let nulls = arrow::buffer::NullBuffer::from(nulls);
 
     (offsets, nulls)
 }
diff --git a/src/arrow_parquet/parquet_writer.rs b/src/arrow_parquet/parquet_writer.rs
index e1b1562..2d7bd8f 100644
--- a/src/arrow_parquet/parquet_writer.rs
+++ b/src/arrow_parquet/parquet_writer.rs
@@ -1,9 +1,6 @@
 use std::sync::Arc;
 
-use arrow::{
-    array::{ArrayRef, RecordBatch, StructArray},
-    datatypes::FieldRef,
-};
+use arrow::array::RecordBatch;
 use arrow_schema::SchemaRef;
 use parquet::{
     arrow::{async_writer::ParquetObjectWriter, AsyncArrowWriter},
@@ -14,13 +11,15 @@ use tokio::runtime::Runtime;
 
 use crate::{
     arrow_parquet::{
-        codec::ParquetCodecOption, pg_to_arrow::collect_attribute_array_from_tuples,
-        schema_visitor::parse_arrow_schema_from_tupledesc, uri_utils::parquet_writer_from_uri,
+        codec::ParquetCodecOption, schema_visitor::parse_arrow_schema_from_tupledesc,
+        uri_utils::parquet_writer_from_uri,
     },
     pgrx_utils::collect_valid_attributes,
     type_compat::{geometry::reset_postgis_context, map::reset_crunchy_map_context},
 };
 
+use super::pg_to_arrow::to_arrow_array;
+
 pub(crate) struct ParquetWriterContext<'a> {
     runtime: Runtime,
     parquet_writer: AsyncArrowWriter<ParquetObjectWriter>,
@@ -67,17 +66,23 @@ impl<'a> ParquetWriterContext<'a> {
         &mut self,
         tuples: Vec<Option<PgHeapTuple<AllocatedByRust>>>,
     ) {
-        pgrx::pg_sys::check_for_interrupts!();
+        let mut record_batches = vec![];
 
-        // collect arrow arrays for each attribute in the tuples
-        let tuple_attribute_arrow_arrays = collect_arrow_attribute_arrays_from_tupledesc(
-            tuples,
-            self.tupledesc.clone(),
-            self.schema.clone(),
-        );
+        for tuple in tuples {
+            pgrx::pg_sys::check_for_interrupts!();
 
-        let struct_array = StructArray::from(tuple_attribute_arrow_arrays);
-        let record_batch = RecordBatch::from(struct_array);
+            if let Some(tuple) = tuple {
+                let record_batch =
+                    pg_tuple_to_record_batch(tuple, &self.tupledesc, self.schema.clone());
+
+                record_batches.push(record_batch);
+            } else {
+                let null_record_batch = RecordBatch::new_empty(self.schema.clone());
+                record_batches.push(null_record_batch);
+            }
+        }
+
+        let record_batch = arrow::compute::concat_batches(&self.schema, &record_batches).unwrap();
 
         let parquet_writer = &mut self.parquet_writer;
 
@@ -92,15 +97,15 @@ impl<'a> ParquetWriterContext<'a> {
     }
 }
 
-fn collect_arrow_attribute_arrays_from_tupledesc(
-    tuples: Vec<Option<PgHeapTuple<AllocatedByRust>>>,
-    tupledesc: PgTupleDesc,
+fn pg_tuple_to_record_batch(
+    tuple: PgHeapTuple<AllocatedByRust>,
+    tupledesc: &PgTupleDesc,
     schema: SchemaRef,
-) -> Vec<(FieldRef, ArrayRef)> {
+) -> RecordBatch {
     let include_generated_columns = true;
-    let attributes = collect_valid_attributes(&tupledesc, include_generated_columns);
+    let attributes = collect_valid_attributes(tupledesc, include_generated_columns);
 
-    let mut tuple_attribute_arrow_arrays = vec![];
+    let mut attribute_arrow_arrays = vec![];
 
     for attribute in attributes {
         let attribute_name = attribute.name();
@@ -110,18 +115,16 @@ fn collect_arrow_attribute_arrays_from_tupledesc(
             .field_with_name(attribute_name)
             .expect("Expected attribute field");
 
-        let (field, array) = collect_attribute_array_from_tuples(
-            &tuples,
+        let (_field, array) = to_arrow_array(
+            &tuple,
             attribute_name,
             attribute_typoid,
             attribute_typmod,
             Arc::new(attribute_field.clone()),
         );
 
-        let tuple_attribute_arrow_array = (field, array);
-
-        tuple_attribute_arrow_arrays.push(tuple_attribute_arrow_array);
+        attribute_arrow_arrays.push(array);
     }
 
-    tuple_attribute_arrow_arrays
+    RecordBatch::try_new(schema, attribute_arrow_arrays).expect("Expected record batch")
 }
diff --git a/src/arrow_parquet/pg_to_arrow.rs b/src/arrow_parquet/pg_to_arrow.rs
index eb59f92..bd14183 100644
--- a/src/arrow_parquet/pg_to_arrow.rs
+++ b/src/arrow_parquet/pg_to_arrow.rs
@@ -47,6 +47,7 @@ pub(crate) trait PgTypeToArrowArray<T: IntoDatum + FromDatum + UnboxDatum> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef);
 }
 
+#[derive(Clone)]
 pub(crate) struct PgToArrowPerAttributeContext<'a> {
     name: &'a str,
     field: FieldRef,
@@ -79,9 +80,9 @@ impl<'a> PgToArrowPerAttributeContext<'a> {
     }
 }
 
-pub(crate) fn collect_attribute_array_from_tuples<'tup>(
-    tuples: &'tup [Option<PgHeapTuple<'tup, AllocatedByRust>>],
-    attribute_name: &'tup str,
+pub(crate) fn to_arrow_array(
+    tuple: &PgHeapTuple<AllocatedByRust>,
+    attribute_name: &str,
     attribute_typoid: Oid,
     attribute_typmod: i32,
     attribute_field: FieldRef,
@@ -96,7 +97,7 @@ pub(crate) fn collect_attribute_array_from_tuples<'tup>(
             attribute_field,
         );
 
-        collect_array_attribute_array_from_tuples(tuples, attribute_context)
+        to_arrow_list_array(tuple, attribute_context)
     } else {
         let attribute_context = PgToArrowPerAttributeContext::new(
             attribute_name,
@@ -105,338 +106,203 @@ pub(crate) fn collect_attribute_array_from_tuples<'tup>(
             attribute_field,
         );
 
-        collect_nonarray_attribute_array_from_tuples(tuples, attribute_context)
+        to_arrow_primitive_array(tuple, attribute_context)
     }
 }
 
-fn collect_nonarray_attribute_array_from_tuples<'tup>(
-    tuples: &'tup [Option<PgHeapTuple<'tup, AllocatedByRust>>],
-    attribute_context: PgToArrowPerAttributeContext<'tup>,
+fn to_arrow_primitive_array(
+    tuple: &PgHeapTuple<AllocatedByRust>,
+    attribute_context: PgToArrowPerAttributeContext,
 ) -> (FieldRef, ArrayRef) {
     match attribute_context.typoid {
         FLOAT4OID => {
-            let mut attribute_values: Vec<Option<f32>> = Vec::new();
+            let attribute_val: Option<f32> = tuple.get_by_name(attribute_context.name).unwrap();
 
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
-
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         FLOAT8OID => {
-            let mut attribute_values: Vec<Option<f64>> = Vec::new();
-
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
+            let attribute_val: Option<f64> = tuple.get_by_name(attribute_context.name).unwrap();
 
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         INT2OID => {
-            let mut attribute_values: Vec<Option<i16>> = Vec::new();
+            let attribute_val: Option<i16> = tuple.get_by_name(attribute_context.name).unwrap();
 
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
-
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         INT4OID => {
-            let mut attribute_values: Vec<Option<i32>> = Vec::new();
-
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
+            let attribute_val: Option<i32> = tuple.get_by_name(attribute_context.name).unwrap();
 
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         INT8OID => {
-            let mut attribute_values: Vec<Option<i64>> = Vec::new();
+            let attribute_val: Option<i64> = tuple.get_by_name(attribute_context.name).unwrap();
 
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
-
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         NUMERICOID => {
             let precision = extract_precision_from_numeric_typmod(attribute_context.typmod);
+
             if precision > MAX_DECIMAL_PRECISION {
                 reset_fallback_to_text_context(attribute_context.typoid, attribute_context.typmod);
 
-                let mut attribute_values: Vec<Option<FallbackToText>> = Vec::new();
-
-                collect_attribute_array_from_tuples_helper(
-                    tuples,
-                    &attribute_context,
-                    &mut attribute_values,
-                );
+                let attribute_val: Option<FallbackToText> =
+                    tuple.get_by_name(attribute_context.name).unwrap();
 
-                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                let (field, array) = attribute_val.to_arrow_array(attribute_context);
                 (field, array)
             } else {
-                let mut attribute_values: Vec<Option<AnyNumeric>> = Vec::new();
+                let scale = extract_scale_from_numeric_typmod(attribute_context.typmod);
+
+                let attribute_context = attribute_context
+                    .with_scale(scale)
+                    .with_precision(precision);
 
-                collect_attribute_array_from_tuples_helper(
-                    tuples,
-                    &attribute_context,
-                    &mut attribute_values,
-                );
+                let attribute_val: Option<AnyNumeric> =
+                    tuple.get_by_name(attribute_context.name).unwrap();
 
-                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                let (field, array) = attribute_val.to_arrow_array(attribute_context);
                 (field, array)
             }
         }
         BOOLOID => {
-            let mut attribute_values: Vec<Option<bool>> = Vec::new();
-
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
+            let attribute_val: Option<bool> = tuple.get_by_name(attribute_context.name).unwrap();
 
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         DATEOID => {
-            let mut attribute_values: Vec<Option<Date>> = Vec::new();
+            let attribute_val: Option<Date> = tuple.get_by_name(attribute_context.name).unwrap();
 
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
-
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         TIMEOID => {
-            let mut attribute_values: Vec<Option<Time>> = Vec::new();
-
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
+            let attribute_val: Option<Time> = tuple.get_by_name(attribute_context.name).unwrap();
 
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         TIMETZOID => {
-            let mut attribute_values: Vec<Option<TimeWithTimeZone>> = Vec::new();
+            let attribute_val: Option<TimeWithTimeZone> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
-
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         TIMESTAMPOID => {
-            let mut attribute_values: Vec<Option<Timestamp>> = Vec::new();
-
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
+            let attribute_val: Option<Timestamp> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         TIMESTAMPTZOID => {
-            let mut attribute_values: Vec<Option<TimestampWithTimeZone>> = Vec::new();
+            let attribute_val: Option<TimestampWithTimeZone> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
-
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         CHAROID => {
-            let mut attribute_values: Vec<Option<i8>> = Vec::new();
-
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
+            let attribute_val: Option<i8> = tuple.get_by_name(attribute_context.name).unwrap();
 
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         TEXTOID => {
-            let mut attribute_values: Vec<Option<String>> = Vec::new();
-
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
+            let attribute_val: Option<String> = tuple.get_by_name(attribute_context.name).unwrap();
 
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         BYTEAOID => {
-            let mut attribute_values: Vec<Option<&[u8]>> = Vec::new();
+            let attribute_val: Option<&[u8]> = tuple.get_by_name(attribute_context.name).unwrap();
 
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
-
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         OIDOID => {
-            let mut attribute_values: Vec<Option<Oid>> = Vec::new();
-
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
+            let attribute_val: Option<Oid> = tuple.get_by_name(attribute_context.name).unwrap();
 
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         _ => {
             if is_composite_type(attribute_context.typoid) {
-                let mut attribute_values: Vec<Option<PgHeapTuple<AllocatedByRust>>> = Vec::new();
-
-                collect_attribute_array_from_tuples_helper(
-                    tuples,
-                    &attribute_context,
-                    &mut attribute_values,
-                );
+                let attribute_val: Option<PgHeapTuple<AllocatedByRust>> =
+                    tuple.get_by_name(attribute_context.name).unwrap();
 
-                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                let (field, array) = attribute_val.to_arrow_array(attribute_context);
                 (field, array)
             } else if is_crunchy_map_typoid(attribute_context.typoid) {
-                let mut attribute_values: Vec<Option<CrunchyMap>> = Vec::new();
+                let attribute_val: Option<CrunchyMap> =
+                    tuple.get_by_name(attribute_context.name).unwrap();
 
-                collect_attribute_array_from_tuples_helper(
-                    tuples,
-                    &attribute_context,
-                    &mut attribute_values,
-                );
-
-                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                let (field, array) = attribute_val.to_arrow_array(attribute_context);
                 (field, array)
             } else if is_postgis_geometry_typoid(attribute_context.typoid) {
-                let mut attribute_values: Vec<Option<Geometry>> = Vec::new();
-
-                collect_attribute_array_from_tuples_helper(
-                    tuples,
-                    &attribute_context,
-                    &mut attribute_values,
-                );
+                let attribute_val: Option<Geometry> =
+                    tuple.get_by_name(attribute_context.name).unwrap();
 
-                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                let (field, array) = attribute_val.to_arrow_array(attribute_context);
                 (field, array)
             } else {
                 reset_fallback_to_text_context(attribute_context.typoid, attribute_context.typmod);
 
-                let mut attribute_values: Vec<Option<FallbackToText>> = Vec::new();
-
-                collect_attribute_array_from_tuples_helper(
-                    tuples,
-                    &attribute_context,
-                    &mut attribute_values,
-                );
+                let attribute_val: Option<FallbackToText> =
+                    tuple.get_by_name(attribute_context.name).unwrap();
 
-                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                let (field, array) = attribute_val.to_arrow_array(attribute_context);
                 (field, array)
             }
         }
     }
 }
 
-pub(crate) fn collect_array_attribute_array_from_tuples<'tup>(
-    tuples: &'tup [Option<PgHeapTuple<'tup, AllocatedByRust>>],
-    attribute_context: PgToArrowPerAttributeContext<'tup>,
+pub(crate) fn to_arrow_list_array(
+    tuple: &PgHeapTuple<AllocatedByRust>,
+    attribute_context: PgToArrowPerAttributeContext,
 ) -> (FieldRef, ArrayRef) {
     match attribute_context.typoid {
         FLOAT4OID => {
-            let mut attribute_values: Vec<Option<pgrx::Array<f32>>> = Vec::new();
+            let attribute_val: Option<pgrx::Array<f32>> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
-
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         FLOAT8OID => {
-            let mut attribute_values: Vec<Option<pgrx::Array<f64>>> = Vec::new();
-
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
+            let attribute_val: Option<pgrx::Array<f64>> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         INT2OID => {
-            let mut attribute_values: Vec<Option<pgrx::Array<i16>>> = Vec::new();
+            let attribute_val: Option<pgrx::Array<i16>> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
-
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         INT4OID => {
-            let mut attribute_values: Vec<Option<pgrx::Array<i32>>> = Vec::new();
-
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
+            let attribute_val: Option<pgrx::Array<i32>> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         INT8OID => {
-            let mut attribute_values: Vec<Option<pgrx::Array<i64>>> = Vec::new();
+            let attribute_val: Option<pgrx::Array<i64>> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
-
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         NUMERICOID => {
@@ -445,223 +311,123 @@ pub(crate) fn collect_array_attribute_array_from_tuples<'tup>(
             if precision > MAX_DECIMAL_PRECISION {
                 reset_fallback_to_text_context(attribute_context.typoid, attribute_context.typmod);
 
-                let mut attribute_values: Vec<Option<pgrx::Array<FallbackToText>>> = Vec::new();
-
-                collect_attribute_array_from_tuples_helper(
-                    tuples,
-                    &attribute_context,
-                    &mut attribute_values,
-                );
+                let attribute_val: Option<pgrx::Array<FallbackToText>> =
+                    tuple.get_by_name(attribute_context.name).unwrap();
 
-                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                let (field, array) = attribute_val.to_arrow_array(attribute_context);
                 (field, array)
             } else {
-                let mut attribute_values: Vec<Option<pgrx::Array<AnyNumeric>>> = Vec::new();
-
                 let scale = extract_scale_from_numeric_typmod(attribute_context.typmod);
 
                 let attribute_context = attribute_context
                     .with_scale(scale)
                     .with_precision(precision);
 
-                collect_attribute_array_from_tuples_helper(
-                    tuples,
-                    &attribute_context,
-                    &mut attribute_values,
-                );
+                let attribute_val: Option<pgrx::Array<AnyNumeric>> =
+                    tuple.get_by_name(attribute_context.name).unwrap();
 
-                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                let (field, array) = attribute_val.to_arrow_array(attribute_context);
                 (field, array)
             }
         }
         BOOLOID => {
-            let mut attribute_values: Vec<Option<pgrx::Array<bool>>> = Vec::new();
+            let attribute_val: Option<pgrx::Array<bool>> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
-
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         DATEOID => {
-            let mut attribute_values: Vec<Option<pgrx::Array<Date>>> = Vec::new();
-
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
+            let attribute_val: Option<pgrx::Array<Date>> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         TIMEOID => {
-            let mut attribute_values: Vec<Option<pgrx::Array<Time>>> = Vec::new();
-
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
+            let attribute_val: Option<pgrx::Array<Time>> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         TIMETZOID => {
-            let mut attribute_values: Vec<Option<pgrx::Array<TimeWithTimeZone>>> = Vec::new();
+            let attribute_val: Option<pgrx::Array<TimeWithTimeZone>> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
-
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         TIMESTAMPOID => {
-            let mut attribute_values: Vec<Option<pgrx::Array<Timestamp>>> = Vec::new();
-
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
+            let attribute_val: Option<pgrx::Array<Timestamp>> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         TIMESTAMPTZOID => {
-            let mut attribute_values: Vec<Option<pgrx::Array<TimestampWithTimeZone>>> = Vec::new();
-
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
+            let attribute_val: Option<pgrx::Array<TimestampWithTimeZone>> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         CHAROID => {
-            let mut attribute_values: Vec<Option<pgrx::Array<i8>>> = Vec::new();
+            let attribute_val: Option<pgrx::Array<i8>> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
-
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         TEXTOID => {
-            let mut attribute_values: Vec<Option<pgrx::Array<String>>> = Vec::new();
-
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
+            let attribute_val: Option<pgrx::Array<String>> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         BYTEAOID => {
-            let mut attribute_values: Vec<Option<pgrx::Array<&[u8]>>> = Vec::new();
-
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
+            let attribute_val: Option<pgrx::Array<&[u8]>> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         OIDOID => {
-            let mut attribute_values: Vec<Option<pgrx::Array<Oid>>> = Vec::new();
+            let attribute_val: Option<pgrx::Array<Oid>> =
+                tuple.get_by_name(attribute_context.name).unwrap();
 
-            collect_attribute_array_from_tuples_helper(
-                tuples,
-                &attribute_context,
-                &mut attribute_values,
-            );
-
-            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            let (field, array) = attribute_val.to_arrow_array(attribute_context);
             (field, array)
         }
         _ => {
             if is_composite_type(attribute_context.typoid) {
-                let mut attribute_values: Vec<Option<pgrx::Array<PgHeapTuple<AllocatedByRust>>>> =
-                    Vec::new();
-
-                collect_attribute_array_from_tuples_helper(
-                    tuples,
-                    &attribute_context,
-                    &mut attribute_values,
-                );
+                let attribute_val: Option<pgrx::Array<PgHeapTuple<AllocatedByRust>>> =
+                    tuple.get_by_name(attribute_context.name).unwrap();
 
-                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                let (field, array) = attribute_val.to_arrow_array(attribute_context);
                 (field, array)
             } else if is_crunchy_map_typoid(attribute_context.typoid) {
-                let mut attribute_values: Vec<Option<pgrx::Array<CrunchyMap>>> = Vec::new();
-
-                collect_attribute_array_from_tuples_helper(
-                    tuples,
-                    &attribute_context,
-                    &mut attribute_values,
-                );
+                let attribute_val: Option<pgrx::Array<CrunchyMap>> =
+                    tuple.get_by_name(attribute_context.name).unwrap();
 
-                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                let (field, array) = attribute_val.to_arrow_array(attribute_context);
                 (field, array)
             } else if is_postgis_geometry_typoid(attribute_context.typoid) {
-                let mut attribute_values: Vec<Option<pgrx::Array<Geometry>>> = Vec::new();
+                let attribute_val: Option<pgrx::Array<Geometry>> =
+                    tuple.get_by_name(attribute_context.name).unwrap();
 
-                collect_attribute_array_from_tuples_helper(
-                    tuples,
-                    &attribute_context,
-                    &mut attribute_values,
-                );
-
-                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                let (field, array) = attribute_val.to_arrow_array(attribute_context);
                 (field, array)
             } else {
                 reset_fallback_to_text_context(attribute_context.typoid, attribute_context.typmod);
 
-                let mut attribute_values: Vec<Option<pgrx::Array<FallbackToText>>> = Vec::new();
-
-                collect_attribute_array_from_tuples_helper(
-                    tuples,
-                    &attribute_context,
-                    &mut attribute_values,
-                );
+                let attribute_val: Option<pgrx::Array<FallbackToText>> =
+                    tuple.get_by_name(attribute_context.name).unwrap();
 
-                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                let (field, array) = attribute_val.to_arrow_array(attribute_context);
                 (field, array)
             }
         }
     }
 }
-
-fn collect_attribute_array_from_tuples_helper<'tup, T>(
-    tuples: &'tup [Option<PgHeapTuple<'tup, AllocatedByRust>>],
-    attribute_context: &PgToArrowPerAttributeContext<'tup>,
-    attribute_values: &mut Vec<Option<T>>,
-) where
-    T: IntoDatum + FromDatum + UnboxDatum<As<'tup> = T> + 'tup,
-{
-    for record in tuples.iter() {
-        pgrx::pg_sys::check_for_interrupts!();
-
-        if let Some(record) = record {
-            let attribute_val: Option<T> = record.get_by_name(attribute_context.name).unwrap();
-            attribute_values.push(attribute_val);
-        } else {
-            attribute_values.push(None);
-        }
-    }
-}
diff --git a/src/arrow_parquet/pg_to_arrow/bool.rs b/src/arrow_parquet/pg_to_arrow/bool.rs
index af1a294..f8704bb 100644
--- a/src/arrow_parquet/pg_to_arrow/bool.rs
+++ b/src/arrow_parquet/pg_to_arrow/bool.rs
@@ -11,25 +11,25 @@ use crate::arrow_parquet::{arrow_utils::arrow_array_offsets, pg_to_arrow::PgType
 use super::PgToArrowPerAttributeContext;
 
 // Bool
-impl PgTypeToArrowArray<bool> for Vec<Option<bool>> {
+impl PgTypeToArrowArray<bool> for Option<bool> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let bool_array = BooleanArray::from(self);
+        let bool_array = BooleanArray::from(vec![self]);
         (context.field, Arc::new(bool_array))
     }
 }
 
 // Bool[]
-impl PgTypeToArrowArray<pgrx::Array<'_, bool>> for Vec<Option<pgrx::Array<'_, bool>>> {
+impl<'a> PgTypeToArrowArray<pgrx::Array<'a, bool>> for Option<pgrx::Array<'a, bool>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
+        let (offsets, nulls) = arrow_array_offsets(&self);
 
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+        let pg_array = if let Some(pg_array) = self {
+            pg_array.iter().collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
 
-        let bools = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
-        let bool_array = BooleanArray::from(bools);
+        let bool_array = BooleanArray::from(pg_array);
 
         let list_field = context.field;
 
diff --git a/src/arrow_parquet/pg_to_arrow/bytea.rs b/src/arrow_parquet/pg_to_arrow/bytea.rs
index f341874..10980a9 100644
--- a/src/arrow_parquet/pg_to_arrow/bytea.rs
+++ b/src/arrow_parquet/pg_to_arrow/bytea.rs
@@ -11,28 +11,25 @@ use crate::arrow_parquet::{arrow_utils::arrow_array_offsets, pg_to_arrow::PgType
 use super::PgToArrowPerAttributeContext;
 
 // Bytea
-impl PgTypeToArrowArray<&[u8]> for Vec<Option<&[u8]>> {
+impl PgTypeToArrowArray<&[u8]> for Option<&[u8]> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let byte_array = BinaryArray::from(self);
+        let byte_array = BinaryArray::from(vec![self]);
         (context.field, Arc::new(byte_array))
     }
 }
 
 // Bytea[]
-impl PgTypeToArrowArray<pgrx::Array<'_, &[u8]>> for Vec<Option<pgrx::Array<'_, &[u8]>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, &[u8]>> for Option<pgrx::Array<'_, &[u8]>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .iter()
-            .map(|v| {
-                v.as_ref()
-                    .map(|pg_array| pg_array.iter().collect::<Vec<_>>())
-            })
-            .collect::<Vec<_>>();
-
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
-
-        let byteas = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
-        let bytea_array = BinaryArray::from(byteas);
+        let (offsets, nulls) = arrow_array_offsets(&self);
+
+        let pg_array = if let Some(pg_array) = &self {
+            pg_array.iter().collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
+
+        let bytea_array = BinaryArray::from(pg_array);
 
         let list_field = context.field;
 
diff --git a/src/arrow_parquet/pg_to_arrow/char.rs b/src/arrow_parquet/pg_to_arrow/char.rs
index bfe4c8d..1dcaf3e 100644
--- a/src/arrow_parquet/pg_to_arrow/char.rs
+++ b/src/arrow_parquet/pg_to_arrow/char.rs
@@ -11,37 +11,31 @@ use crate::arrow_parquet::{arrow_utils::arrow_array_offsets, pg_to_arrow::PgType
 use super::PgToArrowPerAttributeContext;
 
 // Char
-impl PgTypeToArrowArray<i8> for Vec<Option<i8>> {
+impl PgTypeToArrowArray<i8> for Option<i8> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let chars = self
-            .into_iter()
-            .map(|c| c.map(|c| (c as u8 as char).to_string()))
-            .collect::<Vec<_>>();
+        let char = self.map(|c| (c as u8 as char).to_string());
 
-        let char_array = StringArray::from(chars);
+        let char_array = StringArray::from(vec![char]);
 
         (context.field, Arc::new(char_array))
     }
 }
 
 // "Char"[]
-impl PgTypeToArrowArray<pgrx::Array<'_, i8>> for Vec<Option<pgrx::Array<'_, i8>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, i8>> for Option<pgrx::Array<'_, i8>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
-
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
-
-        let chars = pg_array
-            .into_iter()
-            .flatten()
-            .flatten()
-            .map(|c| c.map(|c| (c as u8 as char).to_string()))
-            .collect::<Vec<_>>();
-
-        let char_array = StringArray::from(chars);
+        let (offsets, nulls) = arrow_array_offsets(&self);
+
+        let pg_array = if let Some(pg_array) = self {
+            pg_array
+                .iter()
+                .map(|c| c.map(|c| (c as u8 as char).to_string()))
+                .collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
+
+        let char_array = StringArray::from(pg_array);
 
         let list_field = context.field;
 
diff --git a/src/arrow_parquet/pg_to_arrow/date.rs b/src/arrow_parquet/pg_to_arrow/date.rs
index a78b4a5..68153c1 100644
--- a/src/arrow_parquet/pg_to_arrow/date.rs
+++ b/src/arrow_parquet/pg_to_arrow/date.rs
@@ -15,37 +15,31 @@ use crate::{
 use super::PgToArrowPerAttributeContext;
 
 // Date
-impl PgTypeToArrowArray<Date> for Vec<Option<Date>> {
+impl PgTypeToArrowArray<Date> for Option<Date> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let dates = self
-            .into_iter()
-            .map(|date| date.and_then(date_to_i32))
-            .collect::<Vec<_>>();
+        let date = self.map(date_to_i32);
 
-        let date_array = Date32Array::from(dates);
+        let date_array = Date32Array::from(vec![date]);
 
         (context.field, Arc::new(date_array))
     }
 }
 
 // Date[]
-impl PgTypeToArrowArray<pgrx::Array<'_, Date>> for Vec<Option<pgrx::Array<'_, Date>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, Date>> for Option<pgrx::Array<'_, Date>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
-
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
-
-        let dates = pg_array
-            .into_iter()
-            .flatten()
-            .flatten()
-            .map(|date| date.and_then(date_to_i32))
-            .collect::<Vec<_>>();
-
-        let date_array = Date32Array::from(dates);
+        let (offsets, nulls) = arrow_array_offsets(&self);
+
+        let pg_array = if let Some(pg_array) = self {
+            pg_array
+                .iter()
+                .map(|date| date.map(date_to_i32))
+                .collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
+
+        let date_array = Date32Array::from(pg_array);
 
         let list_field = context.field;
 
diff --git a/src/arrow_parquet/pg_to_arrow/fallback_to_text.rs b/src/arrow_parquet/pg_to_arrow/fallback_to_text.rs
index 7e8e9c8..f6aa429 100644
--- a/src/arrow_parquet/pg_to_arrow/fallback_to_text.rs
+++ b/src/arrow_parquet/pg_to_arrow/fallback_to_text.rs
@@ -14,14 +14,11 @@ use crate::{
 use super::PgToArrowPerAttributeContext;
 
 // Text representation of any type
-impl PgTypeToArrowArray<FallbackToText> for Vec<Option<FallbackToText>> {
+impl PgTypeToArrowArray<FallbackToText> for Option<FallbackToText> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let texts = self
-            .into_iter()
-            .map(|val| val.map(String::from))
-            .collect::<Vec<_>>();
+        let text = self.map(String::from);
 
-        let text_array = StringArray::from(texts);
+        let text_array = StringArray::from(vec![text]);
 
         (context.field, Arc::new(text_array))
     }
@@ -29,24 +26,21 @@ impl PgTypeToArrowArray<FallbackToText> for Vec<Option<FallbackToText>> {
 
 // Text[] representation of any type
 impl PgTypeToArrowArray<pgrx::Array<'_, FallbackToText>>
-    for Vec<Option<pgrx::Array<'_, FallbackToText>>>
+    for Option<pgrx::Array<'_, FallbackToText>>
 {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
-
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
-
-        let texts = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
-
-        let texts = texts
-            .into_iter()
-            .map(|val| val.map(String::from))
-            .collect::<Vec<_>>();
-
-        let text_array = StringArray::from(texts);
+        let (offsets, nulls) = arrow_array_offsets(&self);
+
+        let pg_array = if let Some(pg_array) = self {
+            pg_array
+                .iter()
+                .map(|f| f.map(String::from))
+                .collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
+
+        let text_array = StringArray::from(pg_array);
 
         let list_field = context.field;
 
diff --git a/src/arrow_parquet/pg_to_arrow/float4.rs b/src/arrow_parquet/pg_to_arrow/float4.rs
index b19567f..796cbb8 100644
--- a/src/arrow_parquet/pg_to_arrow/float4.rs
+++ b/src/arrow_parquet/pg_to_arrow/float4.rs
@@ -11,25 +11,25 @@ use crate::arrow_parquet::{arrow_utils::arrow_array_offsets, pg_to_arrow::PgType
 use super::PgToArrowPerAttributeContext;
 
 // Float32
-impl PgTypeToArrowArray<f32> for Vec<Option<f32>> {
+impl PgTypeToArrowArray<f32> for Option<f32> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let float_array = Float32Array::from(self);
+        let float_array = Float32Array::from(vec![self]);
         (context.field, Arc::new(float_array))
     }
 }
 
 // Float32[]
-impl PgTypeToArrowArray<pgrx::Array<'_, f32>> for Vec<Option<pgrx::Array<'_, f32>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, f32>> for Option<pgrx::Array<'_, f32>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
+        let (offsets, nulls) = arrow_array_offsets(&self);
 
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+        let pg_array = if let Some(pg_array) = self {
+            pg_array.iter().collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
 
-        let floats = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
-        let float_array = Float32Array::from(floats);
+        let float_array = Float32Array::from(pg_array);
 
         let list_field = context.field;
 
diff --git a/src/arrow_parquet/pg_to_arrow/float8.rs b/src/arrow_parquet/pg_to_arrow/float8.rs
index 1d40d57..af1e96e 100644
--- a/src/arrow_parquet/pg_to_arrow/float8.rs
+++ b/src/arrow_parquet/pg_to_arrow/float8.rs
@@ -11,26 +11,25 @@ use crate::arrow_parquet::{arrow_utils::arrow_array_offsets, pg_to_arrow::PgType
 use super::PgToArrowPerAttributeContext;
 
 // Float64
-impl PgTypeToArrowArray<f64> for Vec<Option<f64>> {
+impl PgTypeToArrowArray<f64> for Option<f64> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let double_array = Float64Array::from(self);
+        let double_array = Float64Array::from(vec![self]);
         (context.field, Arc::new(double_array))
     }
 }
 
 // Float64[]
-impl PgTypeToArrowArray<pgrx::Array<'_, f64>> for Vec<Option<pgrx::Array<'_, f64>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, f64>> for Option<pgrx::Array<'_, f64>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
+        let (offsets, nulls) = arrow_array_offsets(&self);
 
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+        let pg_array = if let Some(pg_array) = self {
+            pg_array.iter().collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
 
-        let doubles = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
-
-        let double_array = Float64Array::from(doubles);
+        let double_array = Float64Array::from(pg_array);
 
         let list_field = context.field;
 
diff --git a/src/arrow_parquet/pg_to_arrow/geometry.rs b/src/arrow_parquet/pg_to_arrow/geometry.rs
index 48308b2..f9d05b1 100644
--- a/src/arrow_parquet/pg_to_arrow/geometry.rs
+++ b/src/arrow_parquet/pg_to_arrow/geometry.rs
@@ -14,32 +14,28 @@ use crate::{
 use super::PgToArrowPerAttributeContext;
 
 // Geometry
-impl PgTypeToArrowArray<Geometry> for Vec<Option<Geometry>> {
+impl PgTypeToArrowArray<Geometry> for Option<Geometry> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let wkbs = self
-            .iter()
-            .map(|geometry| geometry.as_deref())
-            .collect::<Vec<_>>();
+        let wkb = self.as_deref();
 
-        let wkb_array = BinaryArray::from(wkbs);
+        let wkb_array = BinaryArray::from(vec![wkb]);
 
         (context.field, Arc::new(wkb_array))
     }
 }
 
 // Geometry[]
-impl PgTypeToArrowArray<pgrx::Array<'_, Geometry>> for Vec<Option<pgrx::Array<'_, Geometry>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, Geometry>> for Option<pgrx::Array<'_, Geometry>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
-
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+        let (offsets, nulls) = arrow_array_offsets(&self);
 
-        let wkbs = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
+        let pg_array = if let Some(pg_array) = self {
+            pg_array.iter().collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
 
-        let wkbs = wkbs
+        let wkbs = pg_array
             .iter()
             .map(|geometry| geometry.as_deref())
             .collect::<Vec<_>>();
diff --git a/src/arrow_parquet/pg_to_arrow/int2.rs b/src/arrow_parquet/pg_to_arrow/int2.rs
index 00627fd..3986d4b 100644
--- a/src/arrow_parquet/pg_to_arrow/int2.rs
+++ b/src/arrow_parquet/pg_to_arrow/int2.rs
@@ -11,26 +11,25 @@ use crate::arrow_parquet::{arrow_utils::arrow_array_offsets, pg_to_arrow::PgType
 use super::PgToArrowPerAttributeContext;
 
 // Int16
-impl PgTypeToArrowArray<i16> for Vec<Option<i16>> {
+impl PgTypeToArrowArray<i16> for Option<i16> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let int16_array = Int16Array::from(self);
+        let int16_array = Int16Array::from(vec![self]);
         (context.field, Arc::new(int16_array))
     }
 }
 
 // Int16[]
-impl PgTypeToArrowArray<pgrx::Array<'_, i16>> for Vec<Option<pgrx::Array<'_, i16>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, i16>> for Option<pgrx::Array<'_, i16>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
+        let (offsets, nulls) = arrow_array_offsets(&self);
 
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+        let pg_array = if let Some(pg_array) = self {
+            pg_array.iter().collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
 
-        let int16s = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
-
-        let int16_array = Int16Array::from(int16s);
+        let int16_array = Int16Array::from(pg_array);
 
         let list_field = context.field;
 
diff --git a/src/arrow_parquet/pg_to_arrow/int4.rs b/src/arrow_parquet/pg_to_arrow/int4.rs
index 9bb4f99..9775b5d 100644
--- a/src/arrow_parquet/pg_to_arrow/int4.rs
+++ b/src/arrow_parquet/pg_to_arrow/int4.rs
@@ -11,26 +11,25 @@ use crate::arrow_parquet::{arrow_utils::arrow_array_offsets, pg_to_arrow::PgType
 use super::PgToArrowPerAttributeContext;
 
 // Int32
-impl PgTypeToArrowArray<i32> for Vec<Option<i32>> {
+impl PgTypeToArrowArray<i32> for Option<i32> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let int32_array = Int32Array::from(self);
+        let int32_array = Int32Array::from(vec![self]);
         (context.field, Arc::new(int32_array))
     }
 }
 
 // Int32[]
-impl PgTypeToArrowArray<pgrx::Array<'_, i32>> for Vec<Option<pgrx::Array<'_, i32>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, i32>> for Option<pgrx::Array<'_, i32>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
+        let (offsets, nulls) = arrow_array_offsets(&self);
 
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+        let pg_array = if let Some(pg_array) = self {
+            pg_array.iter().collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
 
-        let int32s = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
-
-        let int32_array = Int32Array::from(int32s);
+        let int32_array = Int32Array::from(pg_array);
 
         let list_field = context.field;
 
diff --git a/src/arrow_parquet/pg_to_arrow/int8.rs b/src/arrow_parquet/pg_to_arrow/int8.rs
index 10d8b58..5a1919d 100644
--- a/src/arrow_parquet/pg_to_arrow/int8.rs
+++ b/src/arrow_parquet/pg_to_arrow/int8.rs
@@ -11,26 +11,25 @@ use crate::arrow_parquet::{arrow_utils::arrow_array_offsets, pg_to_arrow::PgType
 use super::PgToArrowPerAttributeContext;
 
 // Int64
-impl PgTypeToArrowArray<i64> for Vec<Option<i64>> {
+impl PgTypeToArrowArray<i64> for Option<i64> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let int64_array = Int64Array::from(self);
+        let int64_array = Int64Array::from(vec![self]);
         (context.field, Arc::new(int64_array))
     }
 }
 
 // Int64[]
-impl PgTypeToArrowArray<pgrx::Array<'_, i64>> for Vec<Option<pgrx::Array<'_, i64>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, i64>> for Option<pgrx::Array<'_, i64>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
+        let (offsets, nulls) = arrow_array_offsets(&self);
 
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+        let pg_array = if let Some(pg_array) = self {
+            pg_array.iter().collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
 
-        let int64s = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
-
-        let int64_array = Int64Array::from(int64s);
+        let int64_array = Int64Array::from(pg_array);
 
         let list_field = context.field;
 
diff --git a/src/arrow_parquet/pg_to_arrow/map.rs b/src/arrow_parquet/pg_to_arrow/map.rs
index 2889405..1b6e579 100644
--- a/src/arrow_parquet/pg_to_arrow/map.rs
+++ b/src/arrow_parquet/pg_to_arrow/map.rs
@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use arrow::{
-    array::{make_array, ArrayRef, AsArray, ListArray, MapArray},
+    array::{make_array, new_empty_array, ArrayRef, AsArray, ListArray, MapArray},
     datatypes::FieldRef,
 };
 use arrow_schema::DataType;
@@ -18,9 +18,11 @@ use crate::{
 use super::PgToArrowPerAttributeContext;
 
 // crunchy_map.key_<type1>_val_<type2>
-impl<'b> PgTypeToArrowArray<CrunchyMap<'b>> for Vec<Option<CrunchyMap<'b>>> {
+impl<'b> PgTypeToArrowArray<CrunchyMap<'b>> for Option<CrunchyMap<'b>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (map_offsets, map_nulls) = arrow_map_offsets(&self);
+        let maps = vec![self];
+
+        let (map_offsets, map_nulls) = arrow_map_offsets(&maps);
 
         let map_field = context.field;
 
@@ -29,25 +31,13 @@ impl<'b> PgTypeToArrowArray<CrunchyMap<'b>> for Vec<Option<CrunchyMap<'b>>> {
             _ => panic!("Expected Map field"),
         };
 
-        let mut entries = vec![];
+        let map = maps.into_iter().flatten().next();
 
-        for map in self {
-            if let Some(map) = map {
-                entries.push(Some(map.entries));
-            } else {
-                entries.push(None);
-            }
-        }
-
-        let entries = entries
-            .iter()
-            .map(|v| {
-                v.as_ref()
-                    .map(|pg_array| pg_array.iter().collect::<Vec<_>>())
-            })
-            .collect::<Vec<_>>();
-
-        let entries = entries.into_iter().flatten().flatten().collect::<Vec<_>>();
+        let entries = if let Some(map) = map {
+            Some(map.entries)
+        } else {
+            None
+        };
 
         let entries_typoid = domain_array_base_elem_typoid(context.typoid);
 
@@ -58,7 +48,28 @@ impl<'b> PgTypeToArrowArray<CrunchyMap<'b>> for Vec<Option<CrunchyMap<'b>>> {
             entries_field.clone(),
         );
 
-        let (_, entries_array) = entries.to_arrow_array(entries_context);
+        let entries_array = if let Some(entries) = entries {
+            let mut entries_arrays = vec![];
+
+            for entries in entries.iter() {
+                let (_, entries_array) = entries.to_arrow_array(entries_context.clone());
+                entries_arrays.push(entries_array);
+            }
+
+            let entries_arrays = entries_arrays
+                .iter()
+                .map(|array| array.as_ref())
+                .collect::<Vec<_>>();
+
+            if entries_arrays.is_empty() {
+                new_empty_array(entries_field.data_type())
+            } else {
+                // concatenate entries arrays
+                arrow::compute::concat(&entries_arrays).unwrap()
+            }
+        } else {
+            new_empty_array(entries_field.data_type())
+        };
 
         let entries_array = entries_array.as_struct().to_owned();
 
@@ -76,18 +87,16 @@ impl<'b> PgTypeToArrowArray<CrunchyMap<'b>> for Vec<Option<CrunchyMap<'b>>> {
 
 // crunchy_map.key_<type1>_val_<type2>[]
 impl<'b> PgTypeToArrowArray<pgrx::Array<'_, CrunchyMap<'b>>>
-    for Vec<Option<pgrx::Array<'_, CrunchyMap<'b>>>>
+    for Option<pgrx::Array<'_, CrunchyMap<'b>>>
 {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .iter()
-            .map(|v| {
-                v.as_ref()
-                    .map(|pg_array| pg_array.iter().collect::<Vec<_>>())
-            })
-            .collect::<Vec<_>>();
+        let (list_offsets, list_nulls) = arrow_array_offsets(&self);
 
-        let (list_offsets, list_nulls) = arrow_array_offsets(&pg_array);
+        let pg_array = if let Some(pg_array) = &self {
+            pg_array.iter().collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
 
         let list_field = context.field;
 
@@ -96,9 +105,7 @@ impl<'b> PgTypeToArrowArray<pgrx::Array<'_, CrunchyMap<'b>>>
             _ => panic!("Expected List field"),
         };
 
-        let maps = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
-
-        let (map_offsets, map_nulls) = arrow_map_offsets(&maps);
+        let (map_offsets, map_nulls) = arrow_map_offsets(&pg_array);
 
         let entries_field = match map_field.data_type() {
             DataType::Map(entries_field, false) => entries_field.clone(),
@@ -107,15 +114,15 @@ impl<'b> PgTypeToArrowArray<pgrx::Array<'_, CrunchyMap<'b>>>
 
         let mut entries = vec![];
 
-        for map in maps {
-            if let Some(map) = map {
-                entries.push(Some(map.entries));
+        for map in &pg_array {
+            if let Some(map) = &map {
+                entries.push(Some(map.entries.iter().collect::<Vec<_>>()));
             } else {
                 entries.push(None);
             }
         }
 
-        let entries = entries.into_iter().flatten().map(Some).collect::<Vec<_>>();
+        let entries = entries.into_iter().flatten().flatten().collect::<Vec<_>>();
 
         let entries_typoid = domain_array_base_elem_typoid(context.typoid);
 
@@ -126,7 +133,25 @@ impl<'b> PgTypeToArrowArray<pgrx::Array<'_, CrunchyMap<'b>>>
             entries_field.clone(),
         );
 
-        let (_, entries_array) = entries.to_arrow_array(entries_context);
+        // collect entries arrays
+        let mut entries_arrays = vec![];
+
+        for entries in entries {
+            let (_, entries_array) = entries.to_arrow_array(entries_context.clone());
+            entries_arrays.push(entries_array);
+        }
+
+        let entries_arrays = entries_arrays
+            .iter()
+            .map(|array| array.as_ref())
+            .collect::<Vec<_>>();
+
+        let entries_array = if entries_arrays.is_empty() {
+            new_empty_array(entries_field.data_type())
+        } else {
+            // concatenate entries arrays
+            arrow::compute::concat(&entries_arrays).unwrap()
+        };
 
         let entries_array = entries_array.as_struct().to_owned();
 
diff --git a/src/arrow_parquet/pg_to_arrow/numeric.rs b/src/arrow_parquet/pg_to_arrow/numeric.rs
index 0b34a88..681895b 100644
--- a/src/arrow_parquet/pg_to_arrow/numeric.rs
+++ b/src/arrow_parquet/pg_to_arrow/numeric.rs
@@ -17,17 +17,14 @@ use crate::{
 use super::PgToArrowPerAttributeContext;
 
 // Numeric
-impl PgTypeToArrowArray<AnyNumeric> for Vec<Option<AnyNumeric>> {
+impl PgTypeToArrowArray<AnyNumeric> for Option<AnyNumeric> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
         let precision = extract_precision_from_numeric_typmod(context.typmod);
         let scale = extract_scale_from_numeric_typmod(context.typmod);
 
-        let numerics = self
-            .into_iter()
-            .map(|numeric| numeric.and_then(numeric_to_i128))
-            .collect::<Vec<_>>();
+        let numeric = self.map(numeric_to_i128);
 
-        let numeric_array = Decimal128Array::from(numerics)
+        let numeric_array = Decimal128Array::from(vec![numeric])
             .with_precision_and_scale(precision as _, scale as _)
             .unwrap();
 
@@ -36,26 +33,23 @@ impl PgTypeToArrowArray<AnyNumeric> for Vec<Option<AnyNumeric>> {
 }
 
 // Numeric[]
-impl PgTypeToArrowArray<pgrx::Array<'_, AnyNumeric>> for Vec<Option<pgrx::Array<'_, AnyNumeric>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, AnyNumeric>> for Option<pgrx::Array<'_, AnyNumeric>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
+        let (offsets, nulls) = arrow_array_offsets(&self);
 
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+        let pg_array = if let Some(pg_array) = self {
+            pg_array
+                .iter()
+                .map(|numeric| numeric.map(numeric_to_i128))
+                .collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
 
         let precision = extract_precision_from_numeric_typmod(context.typmod);
         let scale = extract_scale_from_numeric_typmod(context.typmod);
 
-        let numerics = pg_array
-            .into_iter()
-            .flatten()
-            .flatten()
-            .map(|time| time.and_then(numeric_to_i128))
-            .collect::<Vec<_>>();
-
-        let numeric_array = Decimal128Array::from(numerics)
+        let numeric_array = Decimal128Array::from(pg_array)
             .with_precision_and_scale(precision as _, scale as _)
             .unwrap();
 
diff --git a/src/arrow_parquet/pg_to_arrow/oid.rs b/src/arrow_parquet/pg_to_arrow/oid.rs
index 775cfbb..128225d 100644
--- a/src/arrow_parquet/pg_to_arrow/oid.rs
+++ b/src/arrow_parquet/pg_to_arrow/oid.rs
@@ -12,36 +12,31 @@ use crate::arrow_parquet::{arrow_utils::arrow_array_offsets, pg_to_arrow::PgType
 use super::PgToArrowPerAttributeContext;
 
 // Oid
-impl PgTypeToArrowArray<Oid> for Vec<Option<Oid>> {
+impl PgTypeToArrowArray<Oid> for Option<Oid> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let oids = self
-            .into_iter()
-            .map(|x| x.map(|x| x.as_u32()))
-            .collect::<Vec<_>>();
+        let oid = self.map(|oid| oid.as_u32());
 
-        let oid_array = UInt32Array::from(oids);
+        let oid_array = UInt32Array::from(vec![oid]);
 
         (context.field, Arc::new(oid_array))
     }
 }
 
 // Oid[]
-impl PgTypeToArrowArray<pgrx::Array<'_, Oid>> for Vec<Option<pgrx::Array<'_, Oid>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, Oid>> for Option<pgrx::Array<'_, Oid>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
-
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
-
-        let oids = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
-        let oids = oids
-            .into_iter()
-            .map(|x| x.map(|x| x.as_u32()))
-            .collect::<Vec<_>>();
-
-        let oid_array = UInt32Array::from(oids);
+        let (offsets, nulls) = arrow_array_offsets(&self);
+
+        let pg_array = if let Some(pg_array) = self {
+            pg_array
+                .iter()
+                .map(|oid| oid.map(|oid| oid.as_u32()))
+                .collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
+
+        let oid_array = UInt32Array::from(pg_array);
 
         let list_field = context.field;
 
diff --git a/src/arrow_parquet/pg_to_arrow/record.rs b/src/arrow_parquet/pg_to_arrow/record.rs
index bb15ae1..ba4d6e8 100644
--- a/src/arrow_parquet/pg_to_arrow/record.rs
+++ b/src/arrow_parquet/pg_to_arrow/record.rs
@@ -1,6 +1,5 @@
 use arrow::{
-    array::{make_array, ArrayRef, ListArray, StructArray},
-    buffer::{BooleanBuffer, NullBuffer},
+    array::{make_array, new_empty_array, ArrayRef, ListArray, StructArray},
     datatypes::FieldRef,
 };
 use arrow_schema::DataType;
@@ -11,11 +10,11 @@ use crate::{
     pgrx_utils::{collect_valid_attributes, tuple_desc},
 };
 
-use super::{collect_attribute_array_from_tuples, PgToArrowPerAttributeContext};
+use super::{to_arrow_array, PgToArrowPerAttributeContext};
 
 // PgHeapTuple
 impl PgTypeToArrowArray<PgHeapTuple<'_, AllocatedByRust>>
-    for Vec<Option<PgHeapTuple<'_, AllocatedByRust>>>
+    for Option<PgHeapTuple<'_, AllocatedByRust>>
 {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
         let struct_field = context.field;
@@ -30,57 +29,52 @@ impl PgTypeToArrowArray<PgHeapTuple<'_, AllocatedByRust>>
         let include_generated_columns = true;
         let attributes = collect_valid_attributes(&tupledesc, include_generated_columns);
 
-        let mut struct_attribute_arrays = vec![];
+        if let Some(tuple) = &self {
+            let mut attribute_arrow_arrays = vec![];
 
-        let tuples = self;
+            for attribute in attributes {
+                let attribute_name = attribute.name();
+                let attribute_typoid = attribute.type_oid().value();
+                let attribute_typmod = attribute.type_mod();
 
-        for attribute in attributes {
-            let attribute_name = attribute.name();
-            let attribute_typoid = attribute.type_oid().value();
-            let attribute_typmod = attribute.type_mod();
+                let attribute_field = fields
+                    .iter()
+                    .find(|field| field.name() == attribute_name)
+                    .expect("Expected attribute field");
 
-            let attribute_field = fields
-                .iter()
-                .find(|field| field.name() == attribute_name)
-                .expect("Expected attribute field");
+                let (_field, array) = to_arrow_array(
+                    tuple,
+                    attribute_name,
+                    attribute_typoid,
+                    attribute_typmod,
+                    attribute_field.clone(),
+                );
 
-            let (_, array) = collect_attribute_array_from_tuples(
-                &tuples,
-                attribute_name,
-                attribute_typoid,
-                attribute_typmod,
-                attribute_field.clone(),
-            );
+                attribute_arrow_arrays.push(array);
+            }
 
-            struct_attribute_arrays.push(array);
-        }
-
-        // determines which structs in the array are null
-        let is_null_buffer =
-            BooleanBuffer::collect_bool(tuples.len(), |idx| tuples.get(idx).unwrap().is_some());
-        let struct_null_buffer = NullBuffer::new(is_null_buffer);
+            let struct_array = StructArray::new(fields.clone(), attribute_arrow_arrays, None);
 
-        let struct_array =
-            StructArray::new(fields, struct_attribute_arrays, Some(struct_null_buffer));
-
-        (struct_field, make_array(struct_array.into()))
+            (struct_field, make_array(struct_array.into()))
+        } else {
+            let null_array = make_array(StructArray::new_null(fields.clone(), 1).into());
+            (struct_field, null_array)
+        }
     }
 }
 
 // PgHeapTuple[]
 impl PgTypeToArrowArray<pgrx::Array<'_, PgHeapTuple<'_, AllocatedByRust>>>
-    for Vec<Option<pgrx::Array<'_, PgHeapTuple<'_, AllocatedByRust>>>>
+    for Option<pgrx::Array<'_, PgHeapTuple<'_, AllocatedByRust>>>
 {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .iter()
-            .map(|v| {
-                v.as_ref()
-                    .map(|pg_array| pg_array.iter().collect::<Vec<_>>())
-            })
-            .collect::<Vec<_>>();
+        let (offsets, nulls) = arrow_array_offsets(&self);
 
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+        let pg_array = if let Some(pg_array) = &self {
+            pg_array.iter().collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
 
         let list_field = context.field;
 
@@ -89,8 +83,6 @@ impl PgTypeToArrowArray<pgrx::Array<'_, PgHeapTuple<'_, AllocatedByRust>>>
             _ => panic!("Expected List field"),
         };
 
-        let tuples = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
-
         let tuples_context = PgToArrowPerAttributeContext::new(
             context.name,
             context.typoid,
@@ -98,7 +90,25 @@ impl PgTypeToArrowArray<pgrx::Array<'_, PgHeapTuple<'_, AllocatedByRust>>>
             struct_field.clone(),
         );
 
-        let (struct_field, struct_array) = tuples.to_arrow_array(tuples_context);
+        // collect struct arrays
+        let mut struct_arrays = vec![];
+
+        for tuple in pg_array {
+            let (_, struct_array) = tuple.to_arrow_array(tuples_context.clone());
+            struct_arrays.push(struct_array);
+        }
+
+        let struct_arrays = struct_arrays
+            .iter()
+            .map(|array| array.as_ref())
+            .collect::<Vec<_>>();
+
+        let struct_array = if struct_arrays.is_empty() {
+            new_empty_array(struct_field.data_type())
+        } else {
+            // concatenate struct arrays
+            arrow::compute::concat(&struct_arrays).unwrap()
+        };
 
         let list_array = ListArray::new(struct_field, offsets, struct_array, Some(nulls));
 
diff --git a/src/arrow_parquet/pg_to_arrow/text.rs b/src/arrow_parquet/pg_to_arrow/text.rs
index a5e9035..5ec6ce0 100644
--- a/src/arrow_parquet/pg_to_arrow/text.rs
+++ b/src/arrow_parquet/pg_to_arrow/text.rs
@@ -11,26 +11,25 @@ use crate::arrow_parquet::{arrow_utils::arrow_array_offsets, pg_to_arrow::PgType
 use super::PgToArrowPerAttributeContext;
 
 // Text
-impl PgTypeToArrowArray<String> for Vec<Option<String>> {
+impl PgTypeToArrowArray<String> for Option<String> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let text_array = StringArray::from(self);
+        let text_array = StringArray::from(vec![self]);
         (context.field, Arc::new(text_array))
     }
 }
 
 // Text[]
-impl PgTypeToArrowArray<pgrx::Array<'_, String>> for Vec<Option<pgrx::Array<'_, String>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, String>> for Option<pgrx::Array<'_, String>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
+        let (offsets, nulls) = arrow_array_offsets(&self);
 
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+        let pg_array = if let Some(pg_array) = self {
+            pg_array.iter().collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
 
-        let texts = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
-
-        let text_array = StringArray::from(texts);
+        let text_array = StringArray::from(pg_array);
 
         let list_field = context.field;
 
diff --git a/src/arrow_parquet/pg_to_arrow/time.rs b/src/arrow_parquet/pg_to_arrow/time.rs
index 6c31df8..ccc91ef 100644
--- a/src/arrow_parquet/pg_to_arrow/time.rs
+++ b/src/arrow_parquet/pg_to_arrow/time.rs
@@ -15,37 +15,31 @@ use crate::{
 use super::PgToArrowPerAttributeContext;
 
 // Time
-impl PgTypeToArrowArray<Time> for Vec<Option<Time>> {
+impl PgTypeToArrowArray<Time> for Option<Time> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let times = self
-            .into_iter()
-            .map(|time| time.and_then(time_to_i64))
-            .collect::<Vec<_>>();
+        let time = self.map(time_to_i64);
 
-        let time_array = Time64MicrosecondArray::from(times);
+        let time_array = Time64MicrosecondArray::from(vec![time]);
 
         (context.field, Arc::new(time_array))
     }
 }
 
 // Time[]
-impl PgTypeToArrowArray<pgrx::Array<'_, Time>> for Vec<Option<pgrx::Array<'_, Time>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, Time>> for Option<pgrx::Array<'_, Time>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
-
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
-
-        let times = pg_array
-            .into_iter()
-            .flatten()
-            .flatten()
-            .map(|time| time.and_then(time_to_i64))
-            .collect::<Vec<_>>();
-
-        let time_array = Time64MicrosecondArray::from(times);
+        let (offsets, nulls) = arrow_array_offsets(&self);
+
+        let pg_array = if let Some(pg_array) = self {
+            pg_array
+                .iter()
+                .map(|time| time.map(time_to_i64))
+                .collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
+
+        let time_array = Time64MicrosecondArray::from(pg_array);
 
         let list_field = context.field;
 
diff --git a/src/arrow_parquet/pg_to_arrow/timestamp.rs b/src/arrow_parquet/pg_to_arrow/timestamp.rs
index cfadf31..4f9cf09 100644
--- a/src/arrow_parquet/pg_to_arrow/timestamp.rs
+++ b/src/arrow_parquet/pg_to_arrow/timestamp.rs
@@ -15,37 +15,31 @@ use crate::{
 use super::PgToArrowPerAttributeContext;
 
 // Timestamp
-impl PgTypeToArrowArray<Timestamp> for Vec<Option<Timestamp>> {
+impl PgTypeToArrowArray<Timestamp> for Option<Timestamp> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let timestamps = self
-            .into_iter()
-            .map(|timstamp| timstamp.and_then(timestamp_to_i64))
-            .collect::<Vec<_>>();
+        let timestamp = self.map(timestamp_to_i64);
 
-        let timestamp_array = TimestampMicrosecondArray::from(timestamps);
+        let timestamp_array = TimestampMicrosecondArray::from(vec![timestamp]);
 
         (context.field, Arc::new(timestamp_array))
     }
 }
 
 // Timestamp[]
-impl PgTypeToArrowArray<pgrx::Array<'_, Timestamp>> for Vec<Option<pgrx::Array<'_, Timestamp>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, Timestamp>> for Option<pgrx::Array<'_, Timestamp>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
-
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
-
-        let timestamps = pg_array
-            .into_iter()
-            .flatten()
-            .flatten()
-            .map(|timestamp| timestamp.and_then(timestamp_to_i64))
-            .collect::<Vec<_>>();
-
-        let timestamp_array = TimestampMicrosecondArray::from(timestamps);
+        let (offsets, nulls) = arrow_array_offsets(&self);
+
+        let pg_array = if let Some(pg_array) = self {
+            pg_array
+                .iter()
+                .map(|timestamp| timestamp.map(timestamp_to_i64))
+                .collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
+
+        let timestamp_array = TimestampMicrosecondArray::from(pg_array);
 
         let list_field = context.field;
 
diff --git a/src/arrow_parquet/pg_to_arrow/timestamptz.rs b/src/arrow_parquet/pg_to_arrow/timestamptz.rs
index 59d067f..b761438 100644
--- a/src/arrow_parquet/pg_to_arrow/timestamptz.rs
+++ b/src/arrow_parquet/pg_to_arrow/timestamptz.rs
@@ -15,14 +15,12 @@ use crate::{
 use super::PgToArrowPerAttributeContext;
 
 // TimestampTz
-impl PgTypeToArrowArray<TimestampWithTimeZone> for Vec<Option<TimestampWithTimeZone>> {
+impl PgTypeToArrowArray<TimestampWithTimeZone> for Option<TimestampWithTimeZone> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let timestamptzs = self
-            .into_iter()
-            .map(|timestamptz| timestamptz.and_then(timestamptz_to_i64))
-            .collect::<Vec<_>>();
+        let timestamptz = self.map(timestamptz_to_i64);
 
-        let timestamptz_array = TimestampMicrosecondArray::from(timestamptzs).with_timezone_utc();
+        let timestamptz_array =
+            TimestampMicrosecondArray::from(vec![timestamptz]).with_timezone_utc();
 
         (context.field, Arc::new(timestamptz_array))
     }
@@ -30,24 +28,21 @@ impl PgTypeToArrowArray<TimestampWithTimeZone> for Vec<Option<TimestampWithTimeZ
 
 // TimestampTz[]
 impl PgTypeToArrowArray<pgrx::Array<'_, TimestampWithTimeZone>>
-    for Vec<Option<pgrx::Array<'_, TimestampWithTimeZone>>>
+    for Option<pgrx::Array<'_, TimestampWithTimeZone>>
 {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
-
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
-
-        let timestamptzs = pg_array
-            .into_iter()
-            .flatten()
-            .flatten()
-            .map(|timestamptz| timestamptz.and_then(timestamptz_to_i64))
-            .collect::<Vec<_>>();
-
-        let timestamptz_array = TimestampMicrosecondArray::from(timestamptzs).with_timezone_utc();
+        let (offsets, nulls) = arrow_array_offsets(&self);
+
+        let pg_array = if let Some(pg_array) = self {
+            pg_array
+                .iter()
+                .map(|timestamptz| timestamptz.map(timestamptz_to_i64))
+                .collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
+
+        let timestamptz_array = TimestampMicrosecondArray::from(pg_array).with_timezone_utc();
 
         let list_field = context.field;
 
diff --git a/src/arrow_parquet/pg_to_arrow/timetz.rs b/src/arrow_parquet/pg_to_arrow/timetz.rs
index 4fe2e55..7e22c71 100644
--- a/src/arrow_parquet/pg_to_arrow/timetz.rs
+++ b/src/arrow_parquet/pg_to_arrow/timetz.rs
@@ -15,14 +15,11 @@ use crate::{
 use super::PgToArrowPerAttributeContext;
 
 // TimeTz
-impl PgTypeToArrowArray<TimeWithTimeZone> for Vec<Option<TimeWithTimeZone>> {
+impl PgTypeToArrowArray<TimeWithTimeZone> for Option<TimeWithTimeZone> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let timetzs = self
-            .into_iter()
-            .map(|timetz| timetz.and_then(timetz_to_i64))
-            .collect::<Vec<_>>();
+        let timetz = self.map(timetz_to_i64);
 
-        let timetz_array = Time64MicrosecondArray::from(timetzs);
+        let timetz_array = Time64MicrosecondArray::from(vec![timetz]);
 
         (context.field, Arc::new(timetz_array))
     }
@@ -30,24 +27,21 @@ impl PgTypeToArrowArray<TimeWithTimeZone> for Vec<Option<TimeWithTimeZone>> {
 
 // TimeTz[]
 impl PgTypeToArrowArray<pgrx::Array<'_, TimeWithTimeZone>>
-    for Vec<Option<pgrx::Array<'_, TimeWithTimeZone>>>
+    for Option<pgrx::Array<'_, TimeWithTimeZone>>
 {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let pg_array = self
-            .into_iter()
-            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
-            .collect::<Vec<_>>();
-
-        let (offsets, nulls) = arrow_array_offsets(&pg_array);
-
-        let timetzs = pg_array
-            .into_iter()
-            .flatten()
-            .flatten()
-            .map(|timetz| timetz.and_then(timetz_to_i64))
-            .collect::<Vec<_>>();
-
-        let timetz_array = Time64MicrosecondArray::from(timetzs);
+        let (offsets, nulls) = arrow_array_offsets(&self);
+
+        let pg_array = if let Some(pg_array) = self {
+            pg_array
+                .iter()
+                .map(|timetz| timetz.map(timetz_to_i64))
+                .collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
+
+        let timetz_array = Time64MicrosecondArray::from(pg_array);
 
         let list_field = context.field;
 
diff --git a/src/lib.rs b/src/lib.rs
index eca35f8..a6c35ed 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -32,6 +32,7 @@ mod tests {
     use crate::parquet_copy_hook::copy_utils::DEFAULT_ROW_GROUP_SIZE;
     use crate::type_compat::fallback_to_text::FallbackToText;
     use crate::type_compat::geometry::Geometry;
+    use crate::type_compat::map::CrunchyMap;
     use pgrx::pg_sys::Oid;
     use pgrx::{
         composite_type,
@@ -226,6 +227,38 @@ mod tests {
         )
     }
 
+    fn assert_crunchy_int_text_map(expected: Option<CrunchyMap>, actual: Option<CrunchyMap>) {
+        if expected.is_none() {
+            assert!(actual.is_none());
+        } else {
+            assert!(actual.is_some());
+
+            let expected = expected.unwrap().entries;
+            let actual = actual.unwrap().entries;
+
+            for (expected, actual) in expected.iter().zip(actual.iter()) {
+                if expected.is_none() {
+                    assert!(actual.is_none());
+                } else {
+                    assert!(actual.is_some());
+
+                    let expected = expected.unwrap();
+                    let actual = actual.unwrap();
+
+                    let expected_key: Option<i32> = expected.get_by_name("key").unwrap();
+                    let actual_key: Option<i32> = actual.get_by_name("key").unwrap();
+
+                    assert_eq!(expected_key, actual_key);
+
+                    let expected_val: Option<String> = expected.get_by_name("val").unwrap();
+                    let actual_val: Option<String> = actual.get_by_name("val").unwrap();
+
+                    assert_eq!(expected_val, actual_val);
+                }
+            }
+        }
+    }
+
     struct TestResult<T> {
         expected: Vec<(Option<T>,)>,
         result: Vec<(Option<T>,)>,
@@ -266,7 +299,7 @@ mod tests {
     fn test_int2_array() {
         let test_table = TestTable::<Vec<Option<i16>>>::new("int2[]".into());
         test_table
-            .insert("INSERT INTO test_expected (a) VALUES (array[1,2,null]), (null), (array[1]);");
+            .insert("INSERT INTO test_expected (a) VALUES (array[1,2,null]), (null), (array[1]), (array[]::int2[]);");
         test_helper(test_table);
     }
 
@@ -282,7 +315,7 @@ mod tests {
         let test_table: TestTable<Vec<Option<i32>>> =
             TestTable::<Vec<Option<i32>>>::new("int4[]".into());
         test_table
-            .insert("INSERT INTO test_expected (a) VALUES (array[1,2,null]), (null), (array[1]);");
+            .insert("INSERT INTO test_expected (a) VALUES (array[1,2,null]), (null), (array[1]), (array[]::int4[]);");
         test_helper(test_table);
     }
 
@@ -297,7 +330,7 @@ mod tests {
     fn test_int8_array() {
         let test_table = TestTable::<Vec<Option<i64>>>::new("int8[]".into());
         test_table
-            .insert("INSERT INTO test_expected (a) VALUES (array[1,2,null]), (null), (array[1]);");
+            .insert("INSERT INTO test_expected (a) VALUES (array[1,2,null]), (null), (array[1]), (array[]::int8[]);");
         test_helper(test_table);
     }
 
@@ -312,7 +345,7 @@ mod tests {
     fn test_float4_array() {
         let test_table = TestTable::<Vec<Option<f32>>>::new("float4[]".into());
         test_table.insert(
-            "INSERT INTO test_expected (a) VALUES (array[1.123,2.2,null]), (null), (array[1]);",
+            "INSERT INTO test_expected (a) VALUES (array[1.123,2.2,null]), (null), (array[1]), (array[]::float4[]);",
         );
         test_helper(test_table);
     }
@@ -328,7 +361,7 @@ mod tests {
     fn test_float8_array() {
         let test_table = TestTable::<Vec<Option<f64>>>::new("float8[]".into());
         test_table.insert(
-            "INSERT INTO test_expected (a) VALUES (array[1.123,2.2,null]), (null), (array[1]);",
+            "INSERT INTO test_expected (a) VALUES (array[1.123,2.2,null]), (null), (array[1]), (array[]::float8[]);",
         );
         test_helper(test_table);
     }
@@ -343,7 +376,7 @@ mod tests {
     #[pg_test]
     fn test_bool_array() {
         let test_table = TestTable::<Vec<Option<bool>>>::new("bool[]".into());
-        test_table.insert("INSERT INTO test_expected (a) VALUES (array[false,true,false]), (array[true,false,null]), (null);");
+        test_table.insert("INSERT INTO test_expected (a) VALUES (array[false,true,false]), (array[true,false,null]), (null), (array[]::bool[]);");
         test_helper(test_table);
     }
 
@@ -358,7 +391,7 @@ mod tests {
     fn test_text_array() {
         let test_table = TestTable::<Vec<Option<String>>>::new("text[]".into());
         test_table.insert(
-            "INSERT INTO test_expected (a) VALUES (array['asd','efg',null]), (array['e']), (null);",
+            "INSERT INTO test_expected (a) VALUES (array['asd','efg',null]), (array['e']), (null), (array[]::text[]);",
         );
         test_helper(test_table);
     }
@@ -374,7 +407,7 @@ mod tests {
     fn test_varchar_array() {
         let test_table = TestTable::<Vec<Option<FallbackToText>>>::new("varchar[]".into());
         test_table.insert(
-            "INSERT INTO test_expected (a) VALUES (array['asd','efg',null]), (array['e']), (null);",
+            "INSERT INTO test_expected (a) VALUES (array['asd','efg',null]), (array['e']), (null), (array[]::varchar[]);",
         );
         test_helper(test_table);
     }
@@ -390,7 +423,7 @@ mod tests {
     fn test_bpchar_array() {
         let test_table = TestTable::<Vec<Option<FallbackToText>>>::new("bpchar[]".into());
         test_table.insert(
-            "INSERT INTO test_expected (a) VALUES (array['asd','efg',null]), (array['e']), (null);",
+            "INSERT INTO test_expected (a) VALUES (array['asd','efg',null]), (array['e']), (null), (array[]::bpchar[]);",
         );
         test_helper(test_table);
     }
@@ -406,7 +439,7 @@ mod tests {
     fn test_name_array() {
         let test_table = TestTable::<Vec<Option<FallbackToText>>>::new("name[]".into());
         test_table.insert(
-            "INSERT INTO test_expected (a) VALUES (array['asd','efg',null]), (array['e']), (null);",
+            "INSERT INTO test_expected (a) VALUES (array['asd','efg',null]), (array['e']), (null), (array[]::name[]);",
         );
         test_helper(test_table);
     }
@@ -431,7 +464,7 @@ mod tests {
         Spi::run(create_enum_query).unwrap();
 
         let test_table = TestTable::<Vec<Option<FallbackToText>>>::new("color[]".into());
-        test_table.insert("INSERT INTO test_expected (a) VALUES (array['red','blue','green',null]::color[]), (array['blue']::color[]), (null);");
+        test_table.insert("INSERT INTO test_expected (a) VALUES (array['red','blue','green',null]::color[]), (array['blue']::color[]), (null), (array[]::color[]);");
         test_helper(test_table);
 
         let drop_enum_query = "DROP TYPE color CASCADE;";
@@ -463,7 +496,7 @@ mod tests {
     fn test_bit_array() {
         let test_table = TestTable::<Vec<Option<FallbackToText>>>::new("bit[]".into());
         test_table.insert(
-            "INSERT INTO test_expected (a) VALUES (array['1','0','1']::bit[]), (array['1']::bit[]), (null);",
+            "INSERT INTO test_expected (a) VALUES (array['1','0','1']::bit[]), (array['1']::bit[]), (null), (array[]::bit[]);",
         );
         test_helper(test_table);
     }
@@ -497,7 +530,7 @@ mod tests {
     fn test_varbit_array() {
         let test_table = TestTable::<Vec<Option<FallbackToText>>>::new("varbit[]".into());
         test_table.insert(
-            "INSERT INTO test_expected (a) VALUES (array['0101','1','1111110010101',null]::varbit[]), (null);",
+            "INSERT INTO test_expected (a) VALUES (array['0101','1','1111110010101',null]::varbit[]), (null), (array[]::varbit[]);",
         );
         test_helper(test_table);
     }
@@ -513,7 +546,7 @@ mod tests {
     fn test_char_array() {
         let test_table = TestTable::<Vec<Option<i8>>>::new("\"char\"[]".into());
         test_table
-            .insert("INSERT INTO test_expected (a) VALUES (array['a','b','c',null]), (null);");
+            .insert("INSERT INTO test_expected (a) VALUES (array['a','b','c',null]), (null), (array[]::\"char\"[]);");
         test_helper(test_table);
     }
 
@@ -530,7 +563,7 @@ mod tests {
     fn test_bytea_array() {
         let test_table = TestTable::<pgrx::Array<&[u8]>>::new("bytea[]".into());
         test_table.insert(
-            "INSERT INTO test_expected (a) VALUES (array[E'\\\\x010203',E'\\\\x040506',null]::bytea[]), (null);",
+            "INSERT INTO test_expected (a) VALUES (array[E'\\\\x010203',E'\\\\x040506',null]::bytea[]), (null), (array[]::bytea[]);",
         );
         let TestResult { expected, result } = test_common(test_table);
 
@@ -562,10 +595,58 @@ mod tests {
     #[pg_test]
     fn test_oid_array() {
         let test_table = TestTable::<Vec<Option<Oid>>>::new("oid[]".into());
-        test_table.insert("INSERT INTO test_expected (a) VALUES (array[1,2,null]), (null);");
+        test_table.insert(
+            "INSERT INTO test_expected (a) VALUES (array[1,2,null]), (null), (array[]::oid[]);",
+        );
         test_helper(test_table);
     }
 
+    #[pg_test]
+    #[ignore = "enable when we install crunchy_map package on CI"]
+    fn test_crunchy_map() {
+        Spi::run("DROP EXTENSION IF EXISTS crunchy_map; CREATE EXTENSION crunchy_map;").unwrap();
+
+        Spi::run("SELECT crunchy_map.create('int','text');").unwrap();
+
+        let test_table = TestTable::<CrunchyMap>::new("crunchy_map.key_int_val_text".into());
+        test_table.insert("INSERT INTO test_expected (a) VALUES ('{\"(1,)\",\"(,myself)\",\"(3,ddd)\"}'::crunchy_map.key_int_val_text), (NULL);");
+
+        let TestResult { expected, result } = test_common(test_table);
+
+        for ((expected,), (actual,)) in expected.into_iter().zip(result.into_iter()) {
+            assert_crunchy_int_text_map(expected, actual);
+        }
+    }
+
+    #[pg_test]
+    #[ignore = "enable when we install crunchy_map package on CI"]
+    fn test_crunchy_map_array() {
+        Spi::run("DROP EXTENSION IF EXISTS crunchy_map; CREATE EXTENSION crunchy_map;").unwrap();
+
+        Spi::run("SELECT crunchy_map.create('int','text');").unwrap();
+
+        let test_table =
+            TestTable::<Vec<Option<CrunchyMap>>>::new("crunchy_map.key_int_val_text[]".into());
+        test_table.insert("INSERT INTO test_expected (a) VALUES (array['{\"(1,)\",\"(,myself)\",\"(3,ddd)\"}']::crunchy_map.key_int_val_text[]), (NULL), (array[]::crunchy_map.key_int_val_text[]);");
+
+        let TestResult { expected, result } = test_common(test_table);
+
+        for ((expected,), (actual,)) in expected.into_iter().zip(result.into_iter()) {
+            if expected.is_none() {
+                assert!(actual.is_none());
+            } else {
+                assert!(actual.is_some());
+
+                let expected = expected.unwrap();
+                let actual = actual.unwrap();
+
+                for (expected, actual) in expected.into_iter().zip(actual.into_iter()) {
+                    assert_crunchy_int_text_map(expected, actual);
+                }
+            }
+        }
+    }
+
     #[pg_test]
     fn test_date() {
         let test_table = TestTable::<Date>::new("date".into());
@@ -578,7 +659,7 @@ mod tests {
     fn test_date_array() {
         let test_table = TestTable::<Vec<Option<Date>>>::new("date[]".into());
         test_table.insert(
-            "INSERT INTO test_expected (a) VALUES (array['2022-05-01','2022-05-02',null]::date[]), (null);",
+            "INSERT INTO test_expected (a) VALUES (array['2022-05-01','2022-05-02',null]::date[]), (null), (array[]::date[]);",
         );
         test_helper(test_table);
     }
@@ -595,7 +676,7 @@ mod tests {
     fn test_time_array() {
         let test_table = TestTable::<Vec<Option<Time>>>::new("time[]".into());
         test_table.insert(
-            "INSERT INTO test_expected (a) VALUES (array['15:00:00','15:30:12',null]::time[]), (null);",
+            "INSERT INTO test_expected (a) VALUES (array['15:00:00','15:30:12',null]::time[]), (null), (array[]::time[]);",
         );
         test_helper(test_table);
     }
@@ -627,7 +708,7 @@ mod tests {
     fn test_timetz_array() {
         let test_table = TestTable::<Vec<Option<TimeWithTimeZone>>>::new("timetz[]".into());
         test_table.insert(
-            "INSERT INTO test_expected (a) VALUES (array['15:00:00+03','15:30:12-03',null]::timetz[]), (null);",
+            "INSERT INTO test_expected (a) VALUES (array['15:00:00+03','15:30:12-03',null]::timetz[]), (null), (array[]::timetz[]);",
         );
         let TestResult { expected, result } = test_common(test_table);
 
@@ -658,7 +739,7 @@ mod tests {
     #[pg_test]
     fn test_timestamp_array() {
         let test_table = TestTable::<Vec<Option<Timestamp>>>::new("timestamp[]".into());
-        test_table.insert("INSERT INTO test_expected (a) VALUES (array['2022-05-01 15:00:00','2022-05-02 15:30:12',null]::timestamp[]), (null);");
+        test_table.insert("INSERT INTO test_expected (a) VALUES (array['2022-05-01 15:00:00','2022-05-02 15:30:12',null]::timestamp[]), (null), (array[]::timestamp[]);");
         test_helper(test_table);
     }
 
@@ -673,7 +754,7 @@ mod tests {
     fn test_timestamptz_array() {
         let test_table =
             TestTable::<Vec<Option<TimestampWithTimeZone>>>::new("timestamptz[]".into());
-        test_table.insert("INSERT INTO test_expected (a) VALUES (array['2022-05-01 15:00:00+03','2022-05-02 15:30:12-03',null]::timestamptz[]), (null);");
+        test_table.insert("INSERT INTO test_expected (a) VALUES (array['2022-05-01 15:00:00+03','2022-05-02 15:30:12-03',null]::timestamptz[]), (null), (array[]::timestamptz[]);");
         test_helper(test_table);
     }
 
@@ -687,7 +768,7 @@ mod tests {
     #[pg_test]
     fn test_interval_array() {
         let test_table = TestTable::<Vec<Option<FallbackToText>>>::new("interval[]".into());
-        test_table.insert("INSERT INTO test_expected (a) VALUES (array['15 years 10 months 1 day 10:00:00','5 days 4 minutes 10 seconds',null]::interval[]), (null);");
+        test_table.insert("INSERT INTO test_expected (a) VALUES (array['15 years 10 months 1 day 10:00:00','5 days 4 minutes 10 seconds',null]::interval[]), (null), (array[]::interval[]);");
         test_helper(test_table);
     }
 
@@ -701,7 +782,7 @@ mod tests {
     #[pg_test]
     fn test_uuid_array() {
         let test_table = TestTable::<Vec<Option<FallbackToText>>>::new("uuid[]".into());
-        test_table.insert("INSERT INTO test_expected (a) VALUES (array['00000000-0000-0000-0000-000000000001','00000000-0000-0000-0000-000000000002',null]::uuid[]), (null);");
+        test_table.insert("INSERT INTO test_expected (a) VALUES (array['00000000-0000-0000-0000-000000000001','00000000-0000-0000-0000-000000000002',null]::uuid[]), (null), (array[]::uuid[]);");
         test_helper(test_table);
     }
 
@@ -721,7 +802,7 @@ mod tests {
     fn test_json_array() {
         let test_table = TestTable::<Vec<Option<FallbackToText>>>::new("json[]".into())
             .with_order_by_col("a::text[]".into());
-        test_table.insert("INSERT INTO test_expected (a) VALUES (array['{\"a\":\"test_json_1\"}','{\"a\":\"test_json_2\"}',null]::json[]), (null);");
+        test_table.insert("INSERT INTO test_expected (a) VALUES (array['{\"a\":\"test_json_1\"}','{\"a\":\"test_json_2\"}',null]::json[]), (null), (array[]::json[]);");
         let TestResult { expected, result } = test_common(test_table);
 
         for ((expected,), (actual,)) in expected.into_iter().zip(result.into_iter()) {
@@ -744,7 +825,7 @@ mod tests {
     #[pg_test]
     fn test_jsonb_array() {
         let test_table = TestTable::<Vec<Option<FallbackToText>>>::new("jsonb[]".into());
-        test_table.insert("INSERT INTO test_expected (a) VALUES (array['{\"a\":\"test_jsonb_1\"}','{\"a\":\"test_jsonb_2\"}',null]::jsonb[]), (null);");
+        test_table.insert("INSERT INTO test_expected (a) VALUES (array['{\"a\":\"test_jsonb_1\"}','{\"a\":\"test_jsonb_2\"}',null]::jsonb[]), (null), (array[]::jsonb[]);");
         let TestResult { expected, result } = test_common(test_table);
 
         for ((expected,), (actual,)) in expected.into_iter().zip(result.into_iter()) {
@@ -764,7 +845,7 @@ mod tests {
     fn test_numeric_array() {
         let test_table = TestTable::<Vec<Option<AnyNumeric>>>::new("numeric(10,4)[]".into());
         test_table.insert(
-            "INSERT INTO test_expected (a) VALUES (array[0.0,.0,1.,+1.020,-2.12313,.3,4,null]), (null);",
+            "INSERT INTO test_expected (a) VALUES (array[0.0,.0,1.,+1.020,-2.12313,.3,4,null]), (null), (array[]::numeric(10,4)[]);",
         );
         test_helper(test_table);
     }
@@ -782,7 +863,7 @@ mod tests {
     fn test_huge_numeric_array() {
         let test_table = TestTable::<Vec<Option<FallbackToText>>>::new("numeric(100,4)[]".into());
         test_table.insert(
-            "INSERT INTO test_expected (a) VALUES (array[0.0,.0,1.,1.020,2.12313,3,null]), (null);",
+            "INSERT INTO test_expected (a) VALUES (array[0.0,.0,1.,1.020,2.12313,3,null]), (null), (array[]::numeric(100,4)[]);",
         );
         test_helper(test_table);
     }
@@ -806,14 +887,7 @@ mod tests {
         Spi::run(query).unwrap();
 
         let test_table = TestTable::<Vec<Option<Geometry>>>::new("geometry[]".into());
-        test_table.insert("INSERT INTO test_expected (a) VALUES (array[ST_GeomFromText('POINT(1 1)'), ST_GeomFromText('POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))'), null]), (null);");
-        test_helper(test_table);
-    }
-
-    #[pg_test]
-    fn test_empty_array() {
-        let test_table = TestTable::<Vec<Option<i32>>>::new("int4[]".into());
-        test_table.insert("INSERT INTO test_expected (a) VALUES (array[]::int4[]), (null);");
+        test_table.insert("INSERT INTO test_expected (a) VALUES (array[ST_GeomFromText('POINT(1 1)'), ST_GeomFromText('POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))'), null]), (null), (array[]::geometry[]);");
         test_helper(test_table);
     }
 
@@ -824,6 +898,8 @@ mod tests {
         Spi::run("CREATE TABLE dog_owners (owner dog_owner);").unwrap();
 
         Spi::run("INSERT INTO dog_owners VALUES (ROW('Alice', ARRAY[('Buddy', 2)::dog, ('Charlie', 3)::dog], ARRAY[1, 2, 3]));").unwrap();
+        Spi::run("INSERT INTO dog_owners VALUES (ROW('Cathie', ARRAY[]::dog[], ARRAY[4, 5, 6]));")
+            .unwrap();
         Spi::run("INSERT INTO dog_owners VALUES (ROW('Bob', ARRAY[('Daisy', 4)::dog, ('Ella', 5)::dog], ARRAY[4, 5, 6]));").unwrap();
         Spi::run("INSERT INTO dog_owners VALUES (ROW('Cathy', NULL, NULL));").unwrap();
         Spi::run("INSERT INTO dog_owners VALUES (NULL);").unwrap();
@@ -1037,18 +1113,6 @@ mod tests {
         todo!("Implement tests for S3 object store");
     }
 
-    #[pg_test]
-    #[ignore = "not yet implemented"]
-    fn test_crunchy_map() {
-        todo!("Implement tests for Crunchy Map");
-    }
-
-    #[pg_test]
-    #[ignore = "not yet implemented"]
-    fn test_crunchy_map_array() {
-        todo!("Implement tests for Crunchy Map array");
-    }
-
     #[pg_test]
     #[should_panic(expected = "unsupported uri invalid_uri")]
     fn test_invalid_uri() {
diff --git a/src/type_compat/pg_arrow_type_conversions.rs b/src/type_compat/pg_arrow_type_conversions.rs
index e70157e..ab709c1 100644
--- a/src/type_compat/pg_arrow_type_conversions.rs
+++ b/src/type_compat/pg_arrow_type_conversions.rs
@@ -7,7 +7,7 @@ use pgrx::{
 
 pub(crate) const MAX_DECIMAL_PRECISION: usize = 38;
 
-pub(crate) fn date_to_i32(date: Date) -> Option<i32> {
+pub(crate) fn date_to_i32(date: Date) -> i32 {
     // PG epoch is (2000-01-01). Convert it to Unix epoch (1970-01-01). +10957 days
     let adjusted_date: Date = unsafe {
         direct_function_call(pg_sys::date_pli, &[date.into_datum(), 10957.into_datum()]).unwrap()
@@ -15,9 +15,7 @@ pub(crate) fn date_to_i32(date: Date) -> Option<i32> {
 
     let adjusted_date_as_bytes: Vec<u8> =
         unsafe { direct_function_call(pg_sys::date_send, &[adjusted_date.into_datum()]).unwrap() };
-    Some(i32::from_be_bytes(
-        adjusted_date_as_bytes[0..4].try_into().unwrap(),
-    ))
+    i32::from_be_bytes(adjusted_date_as_bytes[0..4].try_into().unwrap())
 }
 
 pub(crate) fn i32_to_date(i32_date: i32) -> Option<Date> {
@@ -26,7 +24,7 @@ pub(crate) fn i32_to_date(i32_date: i32) -> Option<Date> {
     Some(adjusted_date)
 }
 
-pub(crate) fn timestamp_to_i64(timestamp: Timestamp) -> Option<i64> {
+pub(crate) fn timestamp_to_i64(timestamp: Timestamp) -> i64 {
     // PG epoch is (2000-01-01). Convert it to Unix epoch (1970-01-01). +10957 days
     let adjustment_interval = Interval::from_days(10957);
     let adjusted_timestamp: Timestamp = unsafe {
@@ -40,9 +38,7 @@ pub(crate) fn timestamp_to_i64(timestamp: Timestamp) -> Option<i64> {
     let adjusted_timestamp_as_bytes: Vec<u8> = unsafe {
         direct_function_call(pg_sys::time_send, &[adjusted_timestamp.into_datum()]).unwrap()
     };
-    Some(i64::from_be_bytes(
-        adjusted_timestamp_as_bytes[0..8].try_into().unwrap(),
-    ))
+    i64::from_be_bytes(adjusted_timestamp_as_bytes[0..8].try_into().unwrap())
 }
 
 pub(crate) fn i64_to_timestamp(i64_timestamp: i64) -> Option<Timestamp> {
@@ -61,7 +57,7 @@ pub(crate) fn i64_to_timestamp(i64_timestamp: i64) -> Option<Timestamp> {
     Some(adjusted_timestamp)
 }
 
-pub(crate) fn timestamptz_to_i64(timestamptz: TimestampWithTimeZone) -> Option<i64> {
+pub(crate) fn timestamptz_to_i64(timestamptz: TimestampWithTimeZone) -> i64 {
     // PG epoch is (2000-01-01). Convert it to Unix epoch (1970-01-01). +10957 days
     let adjustment_interval = Interval::from_days(10957);
     let adjusted_timestamptz: TimestampWithTimeZone = unsafe {
@@ -79,9 +75,7 @@ pub(crate) fn timestamptz_to_i64(timestamptz: TimestampWithTimeZone) -> Option<i
         )
         .unwrap()
     };
-    Some(i64::from_be_bytes(
-        adjusted_timestamptz_as_bytes[0..8].try_into().unwrap(),
-    ))
+    i64::from_be_bytes(adjusted_timestamptz_as_bytes[0..8].try_into().unwrap())
 }
 
 pub(crate) fn i64_to_timestamptz(i64_timestamptz: i64) -> Option<TimestampWithTimeZone> {
@@ -100,17 +94,17 @@ pub(crate) fn i64_to_timestamptz(i64_timestamptz: i64) -> Option<TimestampWithTi
     Some(adjusted_timestamptz)
 }
 
-pub(crate) fn time_to_i64(time: Time) -> Option<i64> {
+pub(crate) fn time_to_i64(time: Time) -> i64 {
     let time_as_bytes: Vec<u8> =
         unsafe { direct_function_call(pg_sys::time_send, &[time.into_datum()]).unwrap() };
-    Some(i64::from_be_bytes(time_as_bytes[0..8].try_into().unwrap()))
+    i64::from_be_bytes(time_as_bytes[0..8].try_into().unwrap())
 }
 
 pub(crate) fn i64_to_time(i64_time: i64) -> Option<Time> {
     Some(i64_time.try_into().unwrap())
 }
 
-pub(crate) fn timetz_to_i64(timetz: TimeWithTimeZone) -> Option<i64> {
+pub(crate) fn timetz_to_i64(timetz: TimeWithTimeZone) -> i64 {
     let timezone_as_secs: AnyNumeric = unsafe {
         direct_function_call(
             pg_sys::extract_timetz,
@@ -132,16 +126,14 @@ pub(crate) fn timetz_to_i64(timetz: TimeWithTimeZone) -> Option<i64> {
     let adjusted_timetz_as_bytes: Vec<u8> = unsafe {
         direct_function_call(pg_sys::timetz_send, &[adjusted_timetz.into_datum()]).unwrap()
     };
-    Some(i64::from_be_bytes(
-        adjusted_timetz_as_bytes[0..8].try_into().unwrap(),
-    ))
+    i64::from_be_bytes(adjusted_timetz_as_bytes[0..8].try_into().unwrap())
 }
 
 pub(crate) fn i64_to_timetz(i64_timetz: i64) -> Option<TimeWithTimeZone> {
     i64_to_time(i64_timetz).map(|time| time.into())
 }
 
-pub(crate) fn numeric_to_i128(numeric: AnyNumeric) -> Option<i128> {
+pub(crate) fn numeric_to_i128(numeric: AnyNumeric) -> i128 {
     // obtain numeric's string representation
     let numeric_str: &CStr =
         unsafe { direct_function_call(pg_sys::numeric_out, &[numeric.into_datum()]).unwrap() };
@@ -160,7 +152,7 @@ pub(crate) fn numeric_to_i128(numeric: AnyNumeric) -> Option<i128> {
     }
     decimal *= sign;
 
-    Some(decimal)
+    decimal
 }
 
 pub(crate) fn i128_to_numeric(i128_decimal: i128, scale: usize) -> Option<AnyNumeric> {