From b2d459d9b7893283990065bc25da2e1b5503b8d3 Mon Sep 17 00:00:00 2001
From: aykut-bozkurt <51649454+aykut-bozkurt@users.noreply.github.com>
Date: Wed, 11 Sep 2024 14:53:01 +0300
Subject: [PATCH] bump pgrx to 0.12.3 (#20)

---
 .github/workflows/ci.yml                      |   2 +-
 Cargo.toml                                    |  10 +-
 src/arrow_parquet/arrow_to_pg.rs              | 192 ++--
 src/arrow_parquet/arrow_to_pg/date.rs         |   2 +-
 src/arrow_parquet/arrow_to_pg/map.rs          |  25 +-
 src/arrow_parquet/arrow_to_pg/time.rs         |   2 +-
 src/arrow_parquet/arrow_to_pg/timestamp.rs    |   2 +-
 src/arrow_parquet/arrow_to_pg/timestamptz.rs  |   2 +-
 src/arrow_parquet/arrow_to_pg/timetz.rs       |   2 +-
 src/arrow_parquet/parquet_writer.rs           |   8 +-
 src/arrow_parquet/pg_to_arrow.rs              | 846 ++++++++++--------
 src/arrow_parquet/pg_to_arrow/bool.rs         |  11 +-
 src/arrow_parquet/pg_to_arrow/bytea.rs        |  14 +-
 src/arrow_parquet/pg_to_arrow/char.rs         |  11 +-
 src/arrow_parquet/pg_to_arrow/date.rs         |  13 +-
 .../pg_to_arrow/fallback_to_text.rs           |  13 +-
 src/arrow_parquet/pg_to_arrow/float4.rs       |  11 +-
 src/arrow_parquet/pg_to_arrow/float8.rs       |  11 +-
 src/arrow_parquet/pg_to_arrow/geometry.rs     |  11 +-
 src/arrow_parquet/pg_to_arrow/int2.rs         |  11 +-
 src/arrow_parquet/pg_to_arrow/int4.rs         |  11 +-
 src/arrow_parquet/pg_to_arrow/int8.rs         |  11 +-
 src/arrow_parquet/pg_to_arrow/map.rs          |  46 +-
 src/arrow_parquet/pg_to_arrow/numeric.rs      |  11 +-
 src/arrow_parquet/pg_to_arrow/oid.rs          |  11 +-
 src/arrow_parquet/pg_to_arrow/record.rs       |  33 +-
 src/arrow_parquet/pg_to_arrow/text.rs         |  11 +-
 src/arrow_parquet/pg_to_arrow/time.rs         |  13 +-
 src/arrow_parquet/pg_to_arrow/timestamp.rs    |  13 +-
 src/arrow_parquet/pg_to_arrow/timestamptz.rs  |  15 +-
 src/arrow_parquet/pg_to_arrow/timetz.rs       |  15 +-
 src/bin/pgrx_embed.rs                         |   2 +
 src/lib.rs                                    | 151 ++--
 src/parquet_copy_hook/copy_from.rs            |  25 +-
 src/parquet_copy_hook/copy_to.rs              |  16 +-
 .../copy_to_dest_receiver.rs                  |  12 +-
 src/parquet_copy_hook/copy_utils.rs           |  97 +-
 src/parquet_copy_hook/hook.rs                 | 217 +++--
 src/pgrx_missing_declerations.rs              |  52 ++
 src/type_compat/fallback_to_text.rs           |  22 +
 src/type_compat/geometry.rs                   |  31 +-
 src/type_compat/map.rs                        |  39 +-
 src/type_compat/pg_arrow_type_conversions.rs  |  50 +-
 43 files changed, 1132 insertions(+), 971 deletions(-)
 create mode 100644 src/bin/pgrx_embed.rs
 create mode 100644 src/pgrx_missing_declerations.rs

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 6041291..fbeed55 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -47,7 +47,7 @@ jobs:
 
       - name: Install and configure pgrx
         run: |
-          cargo install --locked cargo-pgrx@0.11.4
+          cargo install --locked cargo-pgrx@0.12.3
           cargo pgrx init --pg16 $(which pg_config)
 
       - name: Format and lint
diff --git a/Cargo.toml b/Cargo.toml
index 4639ec4..ba6f061 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -4,7 +4,11 @@ version = "0.1.0"
 edition = "2021"
 
 [lib]
-crate-type = ["cdylib"]
+crate-type = ["cdylib","lib"]
+
+[[bin]]
+name = "pgrx_embed_pg_parquet"
+path = "./src/bin/pgrx_embed.rs"
 
 [features]
 default = ["pg16"]
@@ -26,13 +30,13 @@ parquet = {version = "53", default-features = false, features = [
     "zstd",
     "object_store",
 ]}
-pgrx = "=0.11.4"
+pgrx = "=0.12.3"
 serde = {version = "1", default-features = false}
 serde_json = "1"
 tokio = {version = "1", default-features = false, features = ["rt", "time", "macros"]}
 
 [dev-dependencies]
-pgrx-tests = "=0.11.4"
+pgrx-tests = "=0.12.3"
 
 [profile.dev]
 panic = "unwind"
diff --git a/src/arrow_parquet/arrow_to_pg.rs b/src/arrow_parquet/arrow_to_pg.rs
index 3f28448..608c248 100644
--- a/src/arrow_parquet/arrow_to_pg.rs
+++ b/src/arrow_parquet/arrow_to_pg.rs
@@ -4,13 +4,13 @@ use arrow::array::{
     StructArray, Time64MicrosecondArray, TimestampMicrosecondArray, UInt32Array,
 };
 use pgrx::{
+    datum::{Date, Time, TimeWithTimeZone, Timestamp, TimestampWithTimeZone},
     pg_sys::{
         Datum, Oid, BOOLOID, BYTEAOID, CHAROID, DATEOID, FLOAT4OID, FLOAT8OID, INT2OID, INT4OID,
         INT8OID, NUMERICOID, OIDOID, TEXTOID, TIMEOID, TIMESTAMPOID, TIMESTAMPTZOID, TIMETZOID,
     },
     prelude::PgHeapTuple,
-    AllocatedByRust, AnyNumeric, Date, IntoDatum, PgTupleDesc, Time, TimeWithTimeZone, Timestamp,
-    TimestampWithTimeZone,
+    AllocatedByRust, AnyNumeric, IntoDatum, PgTupleDesc,
 };
 
 use crate::{
@@ -92,35 +92,22 @@ pub(crate) fn to_pg_datum(
     attribute_typoid: Oid,
     attribute_typmod: i32,
 ) -> Option<Datum> {
-    if is_composite_type(attribute_typoid) {
-        let attribute_tupledesc = tuple_desc(attribute_typoid, attribute_typmod);
-
-        let attribute_context =
-            ArrowToPgPerAttributeContext::new(attribute_typoid, attribute_typmod)
-                .with_tupledesc(attribute_tupledesc);
-
-        to_pg_composite_datum(attribute_array.into(), attribute_context)
-    } else if is_array_type(attribute_typoid) {
+    if is_array_type(attribute_typoid) {
         let attribute_element_typoid = array_element_typoid(attribute_typoid);
 
         let attribute_context =
             ArrowToPgPerAttributeContext::new(attribute_element_typoid, attribute_typmod);
 
         to_pg_array_datum(attribute_array.into(), attribute_context)
-    } else if is_crunchy_map_typoid(attribute_typoid) {
-        let attribute_context =
-            ArrowToPgPerAttributeContext::new(attribute_typoid, attribute_typmod);
-
-        to_pg_map_datum(attribute_array.into(), attribute_context)
     } else {
         let attribute_context =
             ArrowToPgPerAttributeContext::new(attribute_typoid, attribute_typmod);
 
-        to_pg_primitive_datum(attribute_array, attribute_context)
+        to_pg_nonarray_datum(attribute_array, attribute_context)
     }
 }
 
-fn to_pg_primitive_datum(
+fn to_pg_nonarray_datum(
     primitive_array: ArrayData,
     attribute_context: ArrowToPgPerAttributeContext,
 ) -> Option<Datum> {
@@ -233,10 +220,41 @@ fn to_pg_primitive_datum(
             val.into_datum()
         }
         _ => {
-            if is_postgis_geometry_typoid(attribute_context.typoid) {
-                to_pg_geometry_datum(primitive_array.into(), attribute_context)
+            if is_composite_type(attribute_context.typoid) {
+                let attribute_tupledesc =
+                    tuple_desc(attribute_context.typoid, attribute_context.typmod);
+                let attribute_context = attribute_context.with_tupledesc(attribute_tupledesc);
+
+                let val = <PgHeapTuple<AllocatedByRust> as ArrowArrayToPgType<
+                    StructArray,
+                    PgHeapTuple<AllocatedByRust>,
+                >>::to_pg_type(primitive_array.into(), attribute_context);
+
+                val.into_datum()
+            } else if is_crunchy_map_typoid(attribute_context.typoid) {
+                let val = <CrunchyMap as ArrowArrayToPgType<MapArray, CrunchyMap>>::to_pg_type(
+                    primitive_array.into(),
+                    attribute_context,
+                );
+
+                val.into_datum()
+            } else if is_postgis_geometry_typoid(attribute_context.typoid) {
+                let val = <Geometry as ArrowArrayToPgType<BinaryArray, Geometry>>::to_pg_type(
+                    primitive_array.into(),
+                    attribute_context,
+                );
+
+                val.into_datum()
             } else {
-                to_pg_fallback_to_text_datum(primitive_array.into(), attribute_context)
+                reset_fallback_to_text_context(attribute_context.typoid, attribute_context.typmod);
+
+                let val =
+                    <FallbackToText as ArrowArrayToPgType<StringArray, FallbackToText>>::to_pg_type(
+                        primitive_array.into(),
+                        attribute_context,
+                    );
+
+                val.into_datum()
             }
         }
     }
@@ -372,120 +390,40 @@ fn to_pg_array_datum(
             if is_composite_type(attribute_context.typoid) {
                 let attribute_tupledesc =
                     tuple_desc(attribute_context.typoid, attribute_context.typmod);
+                let attribute_context = attribute_context.with_tupledesc(attribute_tupledesc);
 
-                to_pg_composite_array_datum(
-                    list_array.into(),
-                    attribute_context.with_tupledesc(attribute_tupledesc),
-                )
-            } else if is_crunchy_map_typoid(attribute_context.typoid) {
-                to_pg_map_array_datum(list_array.into(), attribute_context)
-            } else if is_postgis_geometry_typoid(attribute_context.typoid) {
-                to_pg_geometry_array_datum(list_array.into(), attribute_context)
-            } else {
-                to_pg_fallback_to_text_array_datum(list_array.into(), attribute_context)
-            }
-        }
-    }
-}
-
-fn to_pg_composite_datum(
-    struct_array: StructArray,
-    attribute_context: ArrowToPgPerAttributeContext,
-) -> Option<Datum> {
-    let val = <PgHeapTuple<AllocatedByRust> as ArrowArrayToPgType<
-        StructArray,
-        PgHeapTuple<AllocatedByRust>,
-    >>::to_pg_type(struct_array, attribute_context);
-
-    val.into_datum()
-}
+                let val = <Vec<Option<PgHeapTuple<AllocatedByRust>>> as ArrowArrayToPgType<
+                    StructArray,
+                    Vec<Option<PgHeapTuple<AllocatedByRust>>>,
+                >>::to_pg_type(list_array.into(), attribute_context);
 
-fn to_pg_composite_array_datum(
-    struct_array: StructArray,
-    attribute_context: ArrowToPgPerAttributeContext,
-) -> Option<Datum> {
-    let val = <Vec<Option<PgHeapTuple<AllocatedByRust>>> as ArrowArrayToPgType<
-        StructArray,
-        Vec<Option<PgHeapTuple<AllocatedByRust>>>,
-    >>::to_pg_type(struct_array, attribute_context);
-
-    val.into_datum()
-}
-
-fn to_pg_map_datum(
-    map_array: MapArray,
-    attribute_context: ArrowToPgPerAttributeContext,
-) -> Option<Datum> {
-    let val = <CrunchyMap as ArrowArrayToPgType<MapArray, CrunchyMap>>::to_pg_type(
-        map_array,
-        attribute_context,
-    );
-
-    val.into_datum()
-}
-
-fn to_pg_map_array_datum(
-    map_array: MapArray,
-    attribute_context: ArrowToPgPerAttributeContext,
-) -> Option<Datum> {
-    let val = <Vec<Option<CrunchyMap>> as ArrowArrayToPgType<MapArray, Vec<Option<CrunchyMap>>>>::to_pg_type(
-        map_array,
-        attribute_context,
-    );
-
-    val.into_datum()
-}
-
-fn to_pg_geometry_datum(
-    geometry_array: BinaryArray,
-    attribute_context: ArrowToPgPerAttributeContext,
-) -> Option<Datum> {
-    let val = <Geometry as ArrowArrayToPgType<BinaryArray, Geometry>>::to_pg_type(
-        geometry_array,
-        attribute_context,
-    );
-
-    val.into_datum()
-}
+                val.into_datum()
+            } else if is_crunchy_map_typoid(attribute_context.typoid) {
+                let val = <Vec<Option<CrunchyMap>> as ArrowArrayToPgType<
+                    MapArray,
+                    Vec<Option<CrunchyMap>>,
+                >>::to_pg_type(list_array.into(), attribute_context);
 
-fn to_pg_geometry_array_datum(
-    geometry_array: BinaryArray,
-    attribute_context: ArrowToPgPerAttributeContext,
-) -> Option<Datum> {
-    let val = <Vec<Option<Geometry>> as ArrowArrayToPgType<
+                val.into_datum()
+            } else if is_postgis_geometry_typoid(attribute_context.typoid) {
+                let val = <Vec<Option<Geometry>> as ArrowArrayToPgType<
                     BinaryArray,
                     Vec<Option<Geometry>>,
-                >>::to_pg_type(geometry_array, attribute_context);
-
-    val.into_datum()
-}
-
-fn to_pg_fallback_to_text_datum(
-    text_array: StringArray,
-    attribute_context: ArrowToPgPerAttributeContext,
-) -> Option<Datum> {
-    reset_fallback_to_text_context(attribute_context.typoid, attribute_context.typmod);
+                >>::to_pg_type(list_array.into(), attribute_context);
 
-    let val = <FallbackToText as ArrowArrayToPgType<StringArray, FallbackToText>>::to_pg_type(
-        text_array,
-        attribute_context,
-    );
-
-    val.into_datum()
-}
-
-fn to_pg_fallback_to_text_array_datum(
-    text_array: StringArray,
-    attribute_context: ArrowToPgPerAttributeContext,
-) -> Option<Datum> {
-    reset_fallback_to_text_context(attribute_context.typoid, attribute_context.typmod);
+                val.into_datum()
+            } else {
+                reset_fallback_to_text_context(attribute_context.typoid, attribute_context.typmod);
 
-    let val = <Vec<Option<FallbackToText>> as ArrowArrayToPgType<
-        StringArray,
-        Vec<Option<FallbackToText>>,
-    >>::to_pg_type(text_array, attribute_context);
+                let val = <Vec<Option<FallbackToText>> as ArrowArrayToPgType<
+                    StringArray,
+                    Vec<Option<FallbackToText>>,
+                >>::to_pg_type(list_array.into(), attribute_context);
 
-    val.into_datum()
+                val.into_datum()
+            }
+        }
+    }
 }
 
 fn to_pg_numeric_datum(
diff --git a/src/arrow_parquet/arrow_to_pg/date.rs b/src/arrow_parquet/arrow_to_pg/date.rs
index 2ba6bcc..7505de4 100644
--- a/src/arrow_parquet/arrow_to_pg/date.rs
+++ b/src/arrow_parquet/arrow_to_pg/date.rs
@@ -1,5 +1,5 @@
 use arrow::array::{Array, Date32Array};
-use pgrx::Date;
+use pgrx::datum::Date;
 
 use crate::type_compat::pg_arrow_type_conversions::i32_to_date;
 
diff --git a/src/arrow_parquet/arrow_to_pg/map.rs b/src/arrow_parquet/arrow_to_pg/map.rs
index a76d05b..a24fd73 100644
--- a/src/arrow_parquet/arrow_to_pg/map.rs
+++ b/src/arrow_parquet/arrow_to_pg/map.rs
@@ -1,5 +1,5 @@
 use arrow::array::{Array, MapArray, StructArray};
-use pgrx::{prelude::PgHeapTuple, AllocatedByRust};
+use pgrx::{prelude::PgHeapTuple, AllocatedByRust, FromDatum, IntoDatum};
 
 use crate::{
     pgrx_utils::{domain_array_base_elem_typoid, tuple_desc},
@@ -32,9 +32,14 @@ impl<'b, 'a: 'b> ArrowArrayToPgType<'b, MapArray, CrunchyMap<'a>> for CrunchyMap
             >>::to_pg_type(entries_array, entries_context);
 
             if let Some(entries) = entries {
-                // entries cannot be null if the map is not null (arrow does not allow it)
-                let entries = entries.into_iter().flatten().collect();
-                Some(CrunchyMap { entries })
+                let entries_datum = entries.into_datum();
+
+                if let Some(entries_datum) = entries_datum {
+                    let entries = unsafe { pgrx::Array::from_datum(entries_datum, false).unwrap() };
+                    Some(CrunchyMap { entries })
+                } else {
+                    None
+                }
             } else {
                 None
             }
@@ -67,9 +72,15 @@ impl<'b, 'a: 'b> ArrowArrayToPgType<'b, MapArray, Vec<Option<CrunchyMap<'a>>>>
                 >>::to_pg_type(entries_array, entries_context.clone());
 
                 if let Some(entries) = entries {
-                    // entries cannot be null if the map is not null (arrow does not allow it)
-                    let entries = entries.into_iter().flatten().collect();
-                    maps.push(Some(CrunchyMap { entries }));
+                    let entries_datum = entries.into_datum();
+
+                    if let Some(entries_datum) = entries_datum {
+                        let entries =
+                            unsafe { pgrx::Array::from_datum(entries_datum, false).unwrap() };
+                        maps.push(Some(CrunchyMap { entries }))
+                    } else {
+                        maps.push(None);
+                    }
                 } else {
                     maps.push(None);
                 }
diff --git a/src/arrow_parquet/arrow_to_pg/time.rs b/src/arrow_parquet/arrow_to_pg/time.rs
index cdb87fb..77a6e06 100644
--- a/src/arrow_parquet/arrow_to_pg/time.rs
+++ b/src/arrow_parquet/arrow_to_pg/time.rs
@@ -1,5 +1,5 @@
 use arrow::array::{Array, Time64MicrosecondArray};
-use pgrx::Time;
+use pgrx::datum::Time;
 
 use crate::type_compat::pg_arrow_type_conversions::i64_to_time;
 
diff --git a/src/arrow_parquet/arrow_to_pg/timestamp.rs b/src/arrow_parquet/arrow_to_pg/timestamp.rs
index b26d0e9..01a49da 100644
--- a/src/arrow_parquet/arrow_to_pg/timestamp.rs
+++ b/src/arrow_parquet/arrow_to_pg/timestamp.rs
@@ -1,5 +1,5 @@
 use arrow::array::{Array, TimestampMicrosecondArray};
-use pgrx::Timestamp;
+use pgrx::datum::Timestamp;
 
 use crate::type_compat::pg_arrow_type_conversions::i64_to_timestamp;
 
diff --git a/src/arrow_parquet/arrow_to_pg/timestamptz.rs b/src/arrow_parquet/arrow_to_pg/timestamptz.rs
index cbf3235..f2cc340 100644
--- a/src/arrow_parquet/arrow_to_pg/timestamptz.rs
+++ b/src/arrow_parquet/arrow_to_pg/timestamptz.rs
@@ -1,5 +1,5 @@
 use arrow::array::{Array, TimestampMicrosecondArray};
-use pgrx::TimestampWithTimeZone;
+use pgrx::datum::TimestampWithTimeZone;
 
 use crate::type_compat::pg_arrow_type_conversions::i64_to_timestamptz;
 
diff --git a/src/arrow_parquet/arrow_to_pg/timetz.rs b/src/arrow_parquet/arrow_to_pg/timetz.rs
index 184c688..154f021 100644
--- a/src/arrow_parquet/arrow_to_pg/timetz.rs
+++ b/src/arrow_parquet/arrow_to_pg/timetz.rs
@@ -1,5 +1,5 @@
 use arrow::array::{Array, Time64MicrosecondArray};
-use pgrx::TimeWithTimeZone;
+use pgrx::datum::TimeWithTimeZone;
 
 use crate::type_compat::pg_arrow_type_conversions::i64_to_timetz;
 
diff --git a/src/arrow_parquet/parquet_writer.rs b/src/arrow_parquet/parquet_writer.rs
index 1f067ac..e1b1562 100644
--- a/src/arrow_parquet/parquet_writer.rs
+++ b/src/arrow_parquet/parquet_writer.rs
@@ -102,8 +102,6 @@ fn collect_arrow_attribute_arrays_from_tupledesc(
 
     let mut tuple_attribute_arrow_arrays = vec![];
 
-    let mut tuples = tuples;
-
     for attribute in attributes {
         let attribute_name = attribute.name();
         let attribute_typoid = attribute.type_oid().value();
@@ -112,16 +110,14 @@ fn collect_arrow_attribute_arrays_from_tupledesc(
             .field_with_name(attribute_name)
             .expect("Expected attribute field");
 
-        let (field, array, tups) = collect_attribute_array_from_tuples(
-            tuples,
-            tupledesc.clone(),
+        let (field, array) = collect_attribute_array_from_tuples(
+            &tuples,
             attribute_name,
             attribute_typoid,
             attribute_typmod,
             Arc::new(attribute_field.clone()),
         );
 
-        tuples = tups;
         let tuple_attribute_arrow_array = (field, array);
 
         tuple_attribute_arrow_arrays.push(tuple_attribute_arrow_array);
diff --git a/src/arrow_parquet/pg_to_arrow.rs b/src/arrow_parquet/pg_to_arrow.rs
index a70c2be..eb59f92 100644
--- a/src/arrow_parquet/pg_to_arrow.rs
+++ b/src/arrow_parquet/pg_to_arrow.rs
@@ -1,17 +1,16 @@
 use arrow::{array::ArrayRef, datatypes::FieldRef};
 use pgrx::{
+    datum::{Date, Time, TimeWithTimeZone, Timestamp, TimestampWithTimeZone, UnboxDatum},
     heap_tuple::PgHeapTuple,
     pg_sys::{
-        self, deconstruct_array, heap_getattr, Datum, Oid, BOOLOID, BYTEAOID, CHAROID, DATEOID,
-        FLOAT4OID, FLOAT8OID, INT2OID, INT4OID, INT8OID, NUMERICOID, OIDOID, TEXTOID, TIMEOID,
-        TIMESTAMPOID, TIMESTAMPTZOID, TIMETZOID,
+        Oid, BOOLOID, BYTEAOID, CHAROID, DATEOID, FLOAT4OID, FLOAT8OID, INT2OID, INT4OID, INT8OID,
+        NUMERICOID, OIDOID, TEXTOID, TIMEOID, TIMESTAMPOID, TIMESTAMPTZOID, TIMETZOID,
     },
-    AllocatedByRust, AnyNumeric, Date, FromDatum, IntoDatum, PgBox, PgTupleDesc, Time,
-    TimeWithTimeZone, Timestamp, TimestampWithTimeZone,
+    AllocatedByRust, AnyNumeric, FromDatum, IntoDatum,
 };
 
 use crate::{
-    pgrx_utils::{array_element_typoid, is_array_type, is_composite_type, tuple_desc},
+    pgrx_utils::{array_element_typoid, is_array_type, is_composite_type},
     type_compat::{
         fallback_to_text::{reset_fallback_to_text_context, FallbackToText},
         geometry::{is_postgis_geometry_typoid, Geometry},
@@ -44,7 +43,7 @@ pub(crate) mod timestamp;
 pub(crate) mod timestamptz;
 pub(crate) mod timetz;
 
-pub(crate) trait PgTypeToArrowArray<T: IntoDatum + FromDatum> {
+pub(crate) trait PgTypeToArrowArray<T: IntoDatum + FromDatum + UnboxDatum> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef);
 }
 
@@ -53,7 +52,6 @@ pub(crate) struct PgToArrowPerAttributeContext<'a> {
     field: FieldRef,
     typoid: Oid,
     typmod: i32,
-    tupledesc: Option<PgTupleDesc<'a>>,
     scale: Option<usize>,
     precision: Option<usize>,
 }
@@ -65,17 +63,11 @@ impl<'a> PgToArrowPerAttributeContext<'a> {
             field,
             typoid,
             typmod,
-            tupledesc: None,
             scale: None,
             precision: None,
         }
     }
 
-    fn with_tupledesc(mut self, tupledesc: PgTupleDesc<'a>) -> Self {
-        self.tupledesc = Some(tupledesc);
-        self
-    }
-
     fn with_scale(mut self, scale: usize) -> Self {
         self.scale = Some(scale);
         self
@@ -87,31 +79,14 @@ impl<'a> PgToArrowPerAttributeContext<'a> {
     }
 }
 
-pub(crate) fn collect_attribute_array_from_tuples<'a>(
-    tuples: Vec<Option<PgHeapTuple<'a, AllocatedByRust>>>,
-    tupledesc: PgTupleDesc<'a>,
-    attribute_name: &'a str,
+pub(crate) fn collect_attribute_array_from_tuples<'tup>(
+    tuples: &'tup [Option<PgHeapTuple<'tup, AllocatedByRust>>],
+    attribute_name: &'tup str,
     attribute_typoid: Oid,
     attribute_typmod: i32,
     attribute_field: FieldRef,
-) -> (
-    FieldRef,
-    ArrayRef,
-    Vec<Option<PgHeapTuple<'a, AllocatedByRust>>>,
-) {
-    if is_composite_type(attribute_typoid) {
-        let attribute_tupledesc = tuple_desc(attribute_typoid, attribute_typmod);
-
-        let attribute_context = PgToArrowPerAttributeContext::new(
-            attribute_name,
-            attribute_typoid,
-            attribute_typmod,
-            attribute_field,
-        )
-        .with_tupledesc(attribute_tupledesc);
-
-        collect_tuple_attribute_array_from_tuples_helper(tuples, tupledesc, attribute_context)
-    } else if is_array_type(attribute_typoid) {
+) -> (FieldRef, ArrayRef) {
+    if is_array_type(attribute_typoid) {
         let attribute_element_typoid = array_element_typoid(attribute_typoid);
 
         let attribute_context = PgToArrowPerAttributeContext::new(
@@ -121,19 +96,7 @@ pub(crate) fn collect_attribute_array_from_tuples<'a>(
             attribute_field,
         );
 
-        collect_array_attribute_array_from_tuples(tuples, tupledesc, attribute_context)
-    } else if is_crunchy_map_typoid(attribute_typoid) {
-        let attribute_context = PgToArrowPerAttributeContext::new(
-            attribute_name,
-            attribute_typoid,
-            attribute_typmod,
-            attribute_field,
-        );
-
-        collect_array_attribute_array_from_tuples_helper::<CrunchyMap<'_>>(
-            tuples,
-            attribute_context,
-        )
+        collect_array_attribute_array_from_tuples(tuples, attribute_context)
     } else {
         let attribute_context = PgToArrowPerAttributeContext::new(
             attribute_name,
@@ -142,426 +105,563 @@ pub(crate) fn collect_attribute_array_from_tuples<'a>(
             attribute_field,
         );
 
-        collect_primitive_attribute_array_from_tuples(tuples, attribute_context)
+        collect_nonarray_attribute_array_from_tuples(tuples, attribute_context)
     }
 }
 
-fn collect_primitive_attribute_array_from_tuples<'a>(
-    tuples: Vec<Option<PgHeapTuple<'a, AllocatedByRust>>>,
-    attribute_context: PgToArrowPerAttributeContext<'a>,
-) -> (
-    FieldRef,
-    ArrayRef,
-    Vec<Option<PgHeapTuple<'a, AllocatedByRust>>>,
-) {
+fn collect_nonarray_attribute_array_from_tuples<'tup>(
+    tuples: &'tup [Option<PgHeapTuple<'tup, AllocatedByRust>>],
+    attribute_context: PgToArrowPerAttributeContext<'tup>,
+) -> (FieldRef, ArrayRef) {
     match attribute_context.typoid {
         FLOAT4OID => {
-            collect_array_attribute_array_from_tuples_helper::<f32>(tuples, attribute_context)
+            let mut attribute_values: Vec<Option<f32>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
         }
         FLOAT8OID => {
-            collect_array_attribute_array_from_tuples_helper::<f64>(tuples, attribute_context)
+            let mut attribute_values: Vec<Option<f64>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
         }
         INT2OID => {
-            collect_array_attribute_array_from_tuples_helper::<i16>(tuples, attribute_context)
+            let mut attribute_values: Vec<Option<i16>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
         }
         INT4OID => {
-            collect_array_attribute_array_from_tuples_helper::<i32>(tuples, attribute_context)
+            let mut attribute_values: Vec<Option<i32>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
         }
         INT8OID => {
-            collect_array_attribute_array_from_tuples_helper::<i64>(tuples, attribute_context)
+            let mut attribute_values: Vec<Option<i64>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.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);
-                collect_array_attribute_array_from_tuples_helper::<FallbackToText>(
+
+                let mut attribute_values: Vec<Option<FallbackToText>> = Vec::new();
+
+                collect_attribute_array_from_tuples_helper(
                     tuples,
-                    attribute_context,
-                )
+                    &attribute_context,
+                    &mut attribute_values,
+                );
+
+                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                (field, array)
             } else {
-                collect_array_attribute_array_from_tuples_helper::<AnyNumeric>(
+                let mut attribute_values: Vec<Option<AnyNumeric>> = Vec::new();
+
+                collect_attribute_array_from_tuples_helper(
                     tuples,
-                    attribute_context,
-                )
+                    &attribute_context,
+                    &mut attribute_values,
+                );
+
+                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                (field, array)
             }
         }
         BOOLOID => {
-            collect_array_attribute_array_from_tuples_helper::<bool>(tuples, attribute_context)
+            let mut attribute_values: Vec<Option<bool>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
         }
         DATEOID => {
-            collect_array_attribute_array_from_tuples_helper::<Date>(tuples, attribute_context)
+            let mut attribute_values: Vec<Option<Date>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
         }
         TIMEOID => {
-            collect_array_attribute_array_from_tuples_helper::<Time>(tuples, attribute_context)
+            let mut attribute_values: Vec<Option<Time>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
+        }
+        TIMETZOID => {
+            let mut attribute_values: Vec<Option<TimeWithTimeZone>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
         }
-        TIMETZOID => collect_array_attribute_array_from_tuples_helper::<TimeWithTimeZone>(
-            tuples,
-            attribute_context,
-        ),
         TIMESTAMPOID => {
-            collect_array_attribute_array_from_tuples_helper::<Timestamp>(tuples, attribute_context)
+            let mut attribute_values: Vec<Option<Timestamp>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
         }
         TIMESTAMPTZOID => {
-            collect_array_attribute_array_from_tuples_helper::<TimestampWithTimeZone>(
+            let mut attribute_values: Vec<Option<TimestampWithTimeZone>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
                 tuples,
-                attribute_context,
-            )
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
         }
         CHAROID => {
-            collect_array_attribute_array_from_tuples_helper::<i8>(tuples, attribute_context)
+            let mut attribute_values: Vec<Option<i8>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
         }
         TEXTOID => {
-            collect_array_attribute_array_from_tuples_helper::<String>(tuples, attribute_context)
+            let mut attribute_values: Vec<Option<String>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
         }
         BYTEAOID => {
-            collect_array_attribute_array_from_tuples_helper::<&[u8]>(tuples, attribute_context)
+            let mut attribute_values: Vec<Option<&[u8]>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
         }
         OIDOID => {
-            collect_array_attribute_array_from_tuples_helper::<Oid>(tuples, attribute_context)
+            let mut attribute_values: Vec<Option<Oid>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
         }
         _ => {
-            if is_postgis_geometry_typoid(attribute_context.typoid) {
-                collect_array_attribute_array_from_tuples_helper::<Geometry>(
+            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,
-                )
+                    &attribute_context,
+                    &mut attribute_values,
+                );
+
+                let (field, array) = attribute_values.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();
+
+                collect_attribute_array_from_tuples_helper(
+                    tuples,
+                    &attribute_context,
+                    &mut attribute_values,
+                );
+
+                let (field, array) = attribute_values.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 (field, array) = attribute_values.to_arrow_array(attribute_context);
+                (field, array)
             } else {
                 reset_fallback_to_text_context(attribute_context.typoid, attribute_context.typmod);
-                collect_array_attribute_array_from_tuples_helper::<FallbackToText>(
+
+                let mut attribute_values: Vec<Option<FallbackToText>> = Vec::new();
+
+                collect_attribute_array_from_tuples_helper(
                     tuples,
-                    attribute_context,
-                )
+                    &attribute_context,
+                    &mut attribute_values,
+                );
+
+                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                (field, array)
             }
         }
     }
 }
 
-pub(crate) fn collect_array_attribute_array_from_tuples<'a>(
-    tuples: Vec<Option<PgHeapTuple<'a, AllocatedByRust>>>,
-    tupledesc: PgTupleDesc<'a>,
-    attribute_context: PgToArrowPerAttributeContext<'a>,
-) -> (
-    FieldRef,
-    ArrayRef,
-    Vec<Option<PgHeapTuple<'a, AllocatedByRust>>>,
-) {
+pub(crate) fn collect_array_attribute_array_from_tuples<'tup>(
+    tuples: &'tup [Option<PgHeapTuple<'tup, AllocatedByRust>>],
+    attribute_context: PgToArrowPerAttributeContext<'tup>,
+) -> (FieldRef, ArrayRef) {
     match attribute_context.typoid {
-        FLOAT4OID => collect_array_attribute_array_from_tuples_helper::<Vec<Option<f32>>>(
-            tuples,
-            attribute_context,
-        ),
-        FLOAT8OID => collect_array_attribute_array_from_tuples_helper::<Vec<Option<f64>>>(
-            tuples,
-            attribute_context,
-        ),
-        INT2OID => collect_array_attribute_array_from_tuples_helper::<Vec<Option<i16>>>(
-            tuples,
-            attribute_context,
-        ),
-        INT4OID => collect_array_attribute_array_from_tuples_helper::<Vec<Option<i32>>>(
-            tuples,
-            attribute_context,
-        ),
-        INT8OID => collect_array_attribute_array_from_tuples_helper::<Vec<Option<i64>>>(
-            tuples,
-            attribute_context,
-        ),
+        FLOAT4OID => {
+            let mut attribute_values: Vec<Option<pgrx::Array<f32>>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.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 (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
+        }
+        INT2OID => {
+            let mut attribute_values: Vec<Option<pgrx::Array<i16>>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.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 (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
+        }
+        INT8OID => {
+            let mut attribute_values: Vec<Option<pgrx::Array<i64>>> = Vec::new();
+
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.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);
 
-                collect_array_attribute_array_from_tuples_helper::<Vec<Option<FallbackToText>>>(
+                let mut attribute_values: Vec<Option<pgrx::Array<FallbackToText>>> = Vec::new();
+
+                collect_attribute_array_from_tuples_helper(
                     tuples,
-                    attribute_context,
-                )
+                    &attribute_context,
+                    &mut attribute_values,
+                );
+
+                let (field, array) = attribute_values.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);
 
-                collect_array_attribute_array_from_tuples_helper::<Vec<Option<AnyNumeric>>>(
-                    tuples,
-                    attribute_context
-                        .with_precision(precision)
-                        .with_scale(scale),
-                )
-            }
-        }
-        BOOLOID => collect_array_attribute_array_from_tuples_helper::<Vec<Option<bool>>>(
-            tuples,
-            attribute_context,
-        ),
-        DATEOID => collect_array_attribute_array_from_tuples_helper::<Vec<Option<Date>>>(
-            tuples,
-            attribute_context,
-        ),
-        TIMEOID => collect_array_attribute_array_from_tuples_helper::<Vec<Option<Time>>>(
-            tuples,
-            attribute_context,
-        ),
-        TIMETZOID => collect_array_attribute_array_from_tuples_helper::<
-            Vec<Option<TimeWithTimeZone>>,
-        >(tuples, attribute_context),
-        TIMESTAMPOID => collect_array_attribute_array_from_tuples_helper::<Vec<Option<Timestamp>>>(
-            tuples,
-            attribute_context,
-        ),
-        TIMESTAMPTZOID => collect_array_attribute_array_from_tuples_helper::<
-            Vec<Option<TimestampWithTimeZone>>,
-        >(tuples, attribute_context),
-        CHAROID => collect_array_attribute_array_from_tuples_helper::<Vec<Option<i8>>>(
-            tuples,
-            attribute_context,
-        ),
-        TEXTOID => collect_array_attribute_array_from_tuples_helper::<Vec<Option<String>>>(
-            tuples,
-            attribute_context,
-        ),
-        BYTEAOID => collect_array_attribute_array_from_tuples_helper::<Vec<Option<&[u8]>>>(
-            tuples,
-            attribute_context,
-        ),
-        OIDOID => collect_array_attribute_array_from_tuples_helper::<Vec<Option<Oid>>>(
-            tuples,
-            attribute_context,
-        ),
-        _ => {
-            if is_composite_type(attribute_context.typoid) {
-                let attribute_tupledesc =
-                    tuple_desc(attribute_context.typoid, attribute_context.typmod);
+                let attribute_context = attribute_context
+                    .with_scale(scale)
+                    .with_precision(precision);
 
-                collect_array_of_tuple_attribute_array_from_tuples_helper(
-                    tuples,
-                    tupledesc,
-                    attribute_context.with_tupledesc(attribute_tupledesc),
-                )
-            } else if is_crunchy_map_typoid(attribute_context.typoid) {
-                collect_array_attribute_array_from_tuples_helper::<Vec<Option<CrunchyMap<'_>>>>(
+                collect_attribute_array_from_tuples_helper(
                     tuples,
-                    attribute_context,
-                )
-            } else if is_postgis_geometry_typoid(attribute_context.typoid) {
-                collect_array_attribute_array_from_tuples_helper::<Vec<Option<Geometry>>>(
-                    tuples,
-                    attribute_context,
-                )
-            } else {
-                reset_fallback_to_text_context(attribute_context.typoid, attribute_context.typmod);
-                collect_array_attribute_array_from_tuples_helper::<Vec<Option<FallbackToText>>>(
-                    tuples,
-                    attribute_context,
-                )
+                    &attribute_context,
+                    &mut attribute_values,
+                );
+
+                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                (field, array)
             }
         }
-    }
-}
+        BOOLOID => {
+            let mut attribute_values: Vec<Option<pgrx::Array<bool>>> = Vec::new();
 
-fn collect_array_attribute_array_from_tuples_helper<'a, T>(
-    tuples: Vec<Option<PgHeapTuple<'a, AllocatedByRust>>>,
-    attribute_context: PgToArrowPerAttributeContext<'a>,
-) -> (
-    FieldRef,
-    ArrayRef,
-    Vec<Option<PgHeapTuple<'a, AllocatedByRust>>>,
-)
-where
-    T: IntoDatum + FromDatum + 'static,
-    Vec<Option<T>>: PgTypeToArrowArray<T>,
-{
-    let mut attribute_values = vec![];
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
 
-    for record in tuples.iter() {
-        pgrx::pg_sys::check_for_interrupts!();
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
+        }
+        DATEOID => {
+            let mut attribute_values: Vec<Option<pgrx::Array<Date>>> = Vec::new();
 
-        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);
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
         }
-    }
+        TIMEOID => {
+            let mut attribute_values: Vec<Option<pgrx::Array<Time>>> = Vec::new();
 
-    let (field, array) = attribute_values.to_arrow_array(attribute_context);
-    (field, array, tuples)
-}
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
 
-fn collect_tuple_attribute_array_from_tuples_helper<'a>(
-    tuples: Vec<Option<PgHeapTuple<'a, AllocatedByRust>>>,
-    tuple_tupledesc: PgTupleDesc<'a>,
-    attribute_context: PgToArrowPerAttributeContext<'a>,
-) -> (
-    FieldRef,
-    ArrayRef,
-    Vec<Option<PgHeapTuple<'a, AllocatedByRust>>>,
-) {
-    let mut attribute_values = vec![];
-
-    let att_tupledesc = attribute_context
-        .tupledesc
-        .clone()
-        .expect("Expected tuple descriptor");
-
-    let attnum = tuple_tupledesc
-        .iter()
-        .find(|attr| attr.name() == attribute_context.name)
-        .unwrap()
-        .attnum;
-
-    let mut tuples_restored = vec![];
-
-    for record in tuples.into_iter() {
-        pgrx::pg_sys::check_for_interrupts!();
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
+        }
+        TIMETZOID => {
+            let mut attribute_values: Vec<Option<pgrx::Array<TimeWithTimeZone>>> = Vec::new();
 
-        if let Some(record) = record {
-            unsafe {
-                let mut isnull = false;
-                let record = record.into_pg();
-                let att_datum =
-                    heap_getattr(record, attnum as _, tuple_tupledesc.as_ptr(), &mut isnull);
-
-                if isnull {
-                    attribute_values.push(None);
-                } else {
-                    let att_val = tuple_from_datum_and_tupdesc(att_datum, att_tupledesc.clone());
-                    attribute_values.push(Some(att_val));
-                }
-
-                let record = PgHeapTuple::from_heap_tuple(tuple_tupledesc.clone(), record);
-                tuples_restored.push(Some(record.into_owned()));
-            }
-        } else {
-            attribute_values.push(None);
-            tuples_restored.push(None);
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
         }
-    }
+        TIMESTAMPOID => {
+            let mut attribute_values: Vec<Option<pgrx::Array<Timestamp>>> = Vec::new();
 
-    let (field, array) = attribute_values.to_arrow_array(attribute_context);
-    (field, array, tuples_restored)
-}
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
 
-fn collect_array_of_tuple_attribute_array_from_tuples_helper<'a>(
-    tuples: Vec<Option<PgHeapTuple<'a, AllocatedByRust>>>,
-    tuple_tupledesc: PgTupleDesc<'a>,
-    attribute_context: PgToArrowPerAttributeContext<'a>,
-) -> (
-    FieldRef,
-    ArrayRef,
-    Vec<Option<PgHeapTuple<'a, AllocatedByRust>>>,
-) {
-    let mut attribute_values = vec![];
-
-    let attnum = tuple_tupledesc
-        .iter()
-        .find(|attr| attr.name() == attribute_context.name)
-        .unwrap()
-        .attnum;
-
-    let att_tupledesc = attribute_context
-        .tupledesc
-        .clone()
-        .expect("Expected tuple descriptor");
-
-    let mut tuples_restored = vec![];
-
-    for record in tuples.into_iter() {
-        pgrx::pg_sys::check_for_interrupts!();
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
+        }
+        TIMESTAMPTZOID => {
+            let mut attribute_values: Vec<Option<pgrx::Array<TimestampWithTimeZone>>> = Vec::new();
 
-        if let Some(record) = record {
-            unsafe {
-                let mut isnull = false;
-                let record = record.into_pg();
-                let att_datum =
-                    heap_getattr(record, attnum as _, tuple_tupledesc.as_ptr(), &mut isnull);
-
-                if isnull {
-                    attribute_values.push(None);
-                } else {
-                    let att_val =
-                        tuple_array_from_datum_and_tupdesc(att_datum, att_tupledesc.clone());
-                    attribute_values.push(Some(att_val));
-                }
-
-                let record = PgHeapTuple::from_heap_tuple(tuple_tupledesc.clone(), record);
-                tuples_restored.push(Some(record.into_owned()));
-            }
-        } else {
-            attribute_values.push(None);
-            tuples_restored.push(None);
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
+
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
         }
-    }
+        CHAROID => {
+            let mut attribute_values: Vec<Option<pgrx::Array<i8>>> = Vec::new();
 
-    let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
 
-    (field, array, tuples_restored)
-}
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
+        }
+        TEXTOID => {
+            let mut attribute_values: Vec<Option<pgrx::Array<String>>> = Vec::new();
 
-fn tuple_from_datum_and_tupdesc(
-    datum: Datum,
-    tupledesc: PgTupleDesc,
-) -> PgHeapTuple<AllocatedByRust> {
-    unsafe {
-        let htup_header = pg_sys::pg_detoast_datum(datum.cast_mut_ptr()) as pg_sys::HeapTupleHeader;
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
 
-        let mut data = PgBox::<pg_sys::HeapTupleData>::alloc0();
-        data.t_len = pgrx::heap_tuple_header_get_datum_length(htup_header) as u32;
-        data.t_data = htup_header;
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
+        }
+        BYTEAOID => {
+            let mut attribute_values: Vec<Option<pgrx::Array<&[u8]>>> = Vec::new();
 
-        PgHeapTuple::from_heap_tuple(tupledesc.clone(), data.as_ptr()).into_owned()
-    }
-}
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
 
-fn tuple_array_from_datum_and_tupdesc(
-    datum: Datum,
-    tupledesc: PgTupleDesc,
-) -> Vec<Option<PgHeapTuple<AllocatedByRust>>> {
-    unsafe {
-        let arraytype = pg_sys::pg_detoast_datum(datum.cast_mut_ptr()) as *mut pg_sys::ArrayType;
+            let (field, array) = attribute_values.to_arrow_array(attribute_context);
+            (field, array)
+        }
+        OIDOID => {
+            let mut attribute_values: Vec<Option<pgrx::Array<Oid>>> = Vec::new();
 
-        let element_oid = tupledesc.oid();
+            collect_attribute_array_from_tuples_helper(
+                tuples,
+                &attribute_context,
+                &mut attribute_values,
+            );
 
-        let mut num_elem = 0;
-        let mut datums = std::ptr::null_mut();
-        let mut nulls = std::ptr::null_mut();
+            let (field, array) = attribute_values.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();
 
-        let mut type_len = 0;
-        let mut type_by_val = false;
-        let mut type_by_align = 0;
+                collect_attribute_array_from_tuples_helper(
+                    tuples,
+                    &attribute_context,
+                    &mut attribute_values,
+                );
 
-        pg_sys::get_typlenbyvalalign(
-            element_oid,
-            &mut type_len,
-            &mut type_by_val,
-            &mut type_by_align,
-        );
+                let (field, array) = attribute_values.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();
 
-        let arr_elem_type = (*arraytype).elemtype;
-        debug_assert!(arr_elem_type == element_oid);
-
-        deconstruct_array(
-            arraytype,
-            element_oid,
-            type_len as _,
-            type_by_val,
-            type_by_align,
-            &mut datums as _,
-            &mut nulls as _,
-            &mut num_elem,
-        );
+                collect_attribute_array_from_tuples_helper(
+                    tuples,
+                    &attribute_context,
+                    &mut attribute_values,
+                );
 
-        let datums = std::slice::from_raw_parts(datums as *const Datum, num_elem as _);
-        let nulls = std::slice::from_raw_parts(nulls as *const bool, num_elem as _);
+                let (field, array) = attribute_values.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 mut tuples = vec![];
+                collect_attribute_array_from_tuples_helper(
+                    tuples,
+                    &attribute_context,
+                    &mut attribute_values,
+                );
 
-        for (datum, isnull) in datums.iter().zip(nulls.iter()) {
-            if *isnull {
-                tuples.push(None);
+                let (field, array) = attribute_values.to_arrow_array(attribute_context);
+                (field, array)
             } else {
-                let tuple = tuple_from_datum_and_tupdesc(datum.to_owned(), tupledesc.clone());
-                tuples.push(Some(tuple));
+                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 (field, array) = attribute_values.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!();
 
-        tuples
+        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 7f7fce2..af1a294 100644
--- a/src/arrow_parquet/pg_to_arrow/bool.rs
+++ b/src/arrow_parquet/pg_to_arrow/bool.rs
@@ -19,11 +19,16 @@ impl PgTypeToArrowArray<bool> for Vec<Option<bool>> {
 }
 
 // Bool[]
-impl PgTypeToArrowArray<Vec<Option<bool>>> for Vec<Option<Vec<Option<bool>>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, bool>> for Vec<Option<pgrx::Array<'_, bool>>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        let pg_array = self
+            .into_iter()
+            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
+            .collect::<Vec<_>>();
 
-        let bools = self.into_iter().flatten().flatten().collect::<Vec<_>>();
+        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+
+        let bools = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
         let bool_array = BooleanArray::from(bools);
 
         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 abe7dd1..f341874 100644
--- a/src/arrow_parquet/pg_to_arrow/bytea.rs
+++ b/src/arrow_parquet/pg_to_arrow/bytea.rs
@@ -19,11 +19,19 @@ impl PgTypeToArrowArray<&[u8]> for Vec<Option<&[u8]>> {
 }
 
 // Bytea[]
-impl PgTypeToArrowArray<Vec<Option<&[u8]>>> for Vec<Option<Vec<Option<&[u8]>>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, &[u8]>> for Vec<Option<pgrx::Array<'_, &[u8]>>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        let pg_array = self
+            .iter()
+            .map(|v| {
+                v.as_ref()
+                    .map(|pg_array| pg_array.iter().collect::<Vec<_>>())
+            })
+            .collect::<Vec<_>>();
 
-        let byteas = self.into_iter().flatten().flatten().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 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 7d335f4..bfe4c8d 100644
--- a/src/arrow_parquet/pg_to_arrow/char.rs
+++ b/src/arrow_parquet/pg_to_arrow/char.rs
@@ -25,11 +25,16 @@ impl PgTypeToArrowArray<i8> for Vec<Option<i8>> {
 }
 
 // "Char"[]
-impl PgTypeToArrowArray<Vec<Option<i8>>> for Vec<Option<Vec<Option<i8>>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, i8>> for Vec<Option<pgrx::Array<'_, i8>>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        let pg_array = self
+            .into_iter()
+            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
+            .collect::<Vec<_>>();
 
-        let chars = self
+        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+
+        let chars = pg_array
             .into_iter()
             .flatten()
             .flatten()
diff --git a/src/arrow_parquet/pg_to_arrow/date.rs b/src/arrow_parquet/pg_to_arrow/date.rs
index af5ad54..a78b4a5 100644
--- a/src/arrow_parquet/pg_to_arrow/date.rs
+++ b/src/arrow_parquet/pg_to_arrow/date.rs
@@ -5,7 +5,7 @@ use arrow::{
     datatypes::FieldRef,
 };
 use arrow_schema::DataType;
-use pgrx::Date;
+use pgrx::datum::Date;
 
 use crate::{
     arrow_parquet::{arrow_utils::arrow_array_offsets, pg_to_arrow::PgTypeToArrowArray},
@@ -29,11 +29,16 @@ impl PgTypeToArrowArray<Date> for Vec<Option<Date>> {
 }
 
 // Date[]
-impl PgTypeToArrowArray<Vec<Option<Date>>> for Vec<Option<Vec<Option<Date>>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, Date>> for Vec<Option<pgrx::Array<'_, Date>>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        let pg_array = self
+            .into_iter()
+            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
+            .collect::<Vec<_>>();
 
-        let dates = self
+        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+
+        let dates = pg_array
             .into_iter()
             .flatten()
             .flatten()
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 c703151..7e8e9c8 100644
--- a/src/arrow_parquet/pg_to_arrow/fallback_to_text.rs
+++ b/src/arrow_parquet/pg_to_arrow/fallback_to_text.rs
@@ -28,11 +28,18 @@ impl PgTypeToArrowArray<FallbackToText> for Vec<Option<FallbackToText>> {
 }
 
 // Text[] representation of any type
-impl PgTypeToArrowArray<Vec<Option<FallbackToText>>> for Vec<Option<Vec<Option<FallbackToText>>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, FallbackToText>>
+    for Vec<Option<pgrx::Array<'_, FallbackToText>>>
+{
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        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 = self.into_iter().flatten().flatten().collect::<Vec<_>>();
+        let texts = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
 
         let texts = texts
             .into_iter()
diff --git a/src/arrow_parquet/pg_to_arrow/float4.rs b/src/arrow_parquet/pg_to_arrow/float4.rs
index 3806eeb..b19567f 100644
--- a/src/arrow_parquet/pg_to_arrow/float4.rs
+++ b/src/arrow_parquet/pg_to_arrow/float4.rs
@@ -19,11 +19,16 @@ impl PgTypeToArrowArray<f32> for Vec<Option<f32>> {
 }
 
 // Float32[]
-impl PgTypeToArrowArray<Vec<Option<f32>>> for Vec<Option<Vec<Option<f32>>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, f32>> for Vec<Option<pgrx::Array<'_, f32>>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        let pg_array = self
+            .into_iter()
+            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
+            .collect::<Vec<_>>();
 
-        let floats = self.into_iter().flatten().flatten().collect::<Vec<_>>();
+        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+
+        let floats = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
         let float_array = Float32Array::from(floats);
 
         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 9da7e0b..1d40d57 100644
--- a/src/arrow_parquet/pg_to_arrow/float8.rs
+++ b/src/arrow_parquet/pg_to_arrow/float8.rs
@@ -19,11 +19,16 @@ impl PgTypeToArrowArray<f64> for Vec<Option<f64>> {
 }
 
 // Float64[]
-impl PgTypeToArrowArray<Vec<Option<f64>>> for Vec<Option<Vec<Option<f64>>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, f64>> for Vec<Option<pgrx::Array<'_, f64>>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        let pg_array = self
+            .into_iter()
+            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
+            .collect::<Vec<_>>();
 
-        let doubles = self.into_iter().flatten().flatten().collect::<Vec<_>>();
+        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+
+        let doubles = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
 
         let double_array = Float64Array::from(doubles);
 
diff --git a/src/arrow_parquet/pg_to_arrow/geometry.rs b/src/arrow_parquet/pg_to_arrow/geometry.rs
index 34e5589..48308b2 100644
--- a/src/arrow_parquet/pg_to_arrow/geometry.rs
+++ b/src/arrow_parquet/pg_to_arrow/geometry.rs
@@ -28,11 +28,16 @@ impl PgTypeToArrowArray<Geometry> for Vec<Option<Geometry>> {
 }
 
 // Geometry[]
-impl PgTypeToArrowArray<Vec<Option<Geometry>>> for Vec<Option<Vec<Option<Geometry>>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, Geometry>> for Vec<Option<pgrx::Array<'_, Geometry>>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        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 wkbs = self.into_iter().flatten().flatten().collect::<Vec<_>>();
+        let wkbs = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
 
         let wkbs = wkbs
             .iter()
diff --git a/src/arrow_parquet/pg_to_arrow/int2.rs b/src/arrow_parquet/pg_to_arrow/int2.rs
index 0cf85a6..00627fd 100644
--- a/src/arrow_parquet/pg_to_arrow/int2.rs
+++ b/src/arrow_parquet/pg_to_arrow/int2.rs
@@ -19,11 +19,16 @@ impl PgTypeToArrowArray<i16> for Vec<Option<i16>> {
 }
 
 // Int16[]
-impl PgTypeToArrowArray<Vec<Option<i16>>> for Vec<Option<Vec<Option<i16>>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, i16>> for Vec<Option<pgrx::Array<'_, i16>>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        let pg_array = self
+            .into_iter()
+            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
+            .collect::<Vec<_>>();
 
-        let int16s = self.into_iter().flatten().flatten().collect::<Vec<_>>();
+        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+
+        let int16s = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
 
         let int16_array = Int16Array::from(int16s);
 
diff --git a/src/arrow_parquet/pg_to_arrow/int4.rs b/src/arrow_parquet/pg_to_arrow/int4.rs
index 4a09f69..9bb4f99 100644
--- a/src/arrow_parquet/pg_to_arrow/int4.rs
+++ b/src/arrow_parquet/pg_to_arrow/int4.rs
@@ -19,11 +19,16 @@ impl PgTypeToArrowArray<i32> for Vec<Option<i32>> {
 }
 
 // Int32[]
-impl PgTypeToArrowArray<Vec<Option<i32>>> for Vec<Option<Vec<Option<i32>>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, i32>> for Vec<Option<pgrx::Array<'_, i32>>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        let pg_array = self
+            .into_iter()
+            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
+            .collect::<Vec<_>>();
 
-        let int32s = self.into_iter().flatten().flatten().collect::<Vec<_>>();
+        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+
+        let int32s = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
 
         let int32_array = Int32Array::from(int32s);
 
diff --git a/src/arrow_parquet/pg_to_arrow/int8.rs b/src/arrow_parquet/pg_to_arrow/int8.rs
index dfe012d..10d8b58 100644
--- a/src/arrow_parquet/pg_to_arrow/int8.rs
+++ b/src/arrow_parquet/pg_to_arrow/int8.rs
@@ -19,11 +19,16 @@ impl PgTypeToArrowArray<i64> for Vec<Option<i64>> {
 }
 
 // Int64[]
-impl PgTypeToArrowArray<Vec<Option<i64>>> for Vec<Option<Vec<Option<i64>>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, i64>> for Vec<Option<pgrx::Array<'_, i64>>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        let pg_array = self
+            .into_iter()
+            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
+            .collect::<Vec<_>>();
 
-        let int64s = self.into_iter().flatten().flatten().collect::<Vec<_>>();
+        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+
+        let int64s = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
 
         let int64_array = Int64Array::from(int64s);
 
diff --git a/src/arrow_parquet/pg_to_arrow/map.rs b/src/arrow_parquet/pg_to_arrow/map.rs
index 4813185..2889405 100644
--- a/src/arrow_parquet/pg_to_arrow/map.rs
+++ b/src/arrow_parquet/pg_to_arrow/map.rs
@@ -11,7 +11,7 @@ use crate::{
         arrow_utils::{arrow_array_offsets, arrow_map_offsets},
         pg_to_arrow::PgTypeToArrowArray,
     },
-    pgrx_utils::{domain_array_base_elem_typoid, tuple_desc},
+    pgrx_utils::domain_array_base_elem_typoid,
     type_compat::map::CrunchyMap,
 };
 
@@ -33,25 +33,30 @@ impl<'b> PgTypeToArrowArray<CrunchyMap<'b>> for Vec<Option<CrunchyMap<'b>>> {
 
         for map in self {
             if let Some(map) = map {
-                entries.push(map.entries);
+                entries.push(Some(map.entries));
             } else {
-                entries.push(vec![]);
+                entries.push(None);
             }
         }
 
-        let entries = entries.into_iter().flatten().map(Some).collect::<Vec<_>>();
+        let entries = entries
+            .iter()
+            .map(|v| {
+                v.as_ref()
+                    .map(|pg_array| pg_array.iter().collect::<Vec<_>>())
+            })
+            .collect::<Vec<_>>();
 
-        let entries_typoid = domain_array_base_elem_typoid(context.typoid);
+        let entries = entries.into_iter().flatten().flatten().collect::<Vec<_>>();
 
-        let entries_tupledesc = tuple_desc(entries_typoid, -1);
+        let entries_typoid = domain_array_base_elem_typoid(context.typoid);
 
         let entries_context = PgToArrowPerAttributeContext::new(
             context.name,
             entries_typoid,
             context.typmod,
             entries_field.clone(),
-        )
-        .with_tupledesc(entries_tupledesc);
+        );
 
         let (_, entries_array) = entries.to_arrow_array(entries_context);
 
@@ -70,11 +75,19 @@ impl<'b> PgTypeToArrowArray<CrunchyMap<'b>> for Vec<Option<CrunchyMap<'b>>> {
 }
 
 // crunchy_map.key_<type1>_val_<type2>[]
-impl<'b> PgTypeToArrowArray<Vec<Option<CrunchyMap<'b>>>>
-    for Vec<Option<Vec<Option<CrunchyMap<'b>>>>>
+impl<'b> PgTypeToArrowArray<pgrx::Array<'_, CrunchyMap<'b>>>
+    for Vec<Option<pgrx::Array<'_, CrunchyMap<'b>>>>
 {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (list_offsets, list_nulls) = arrow_array_offsets(&self);
+        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(&pg_array);
 
         let list_field = context.field;
 
@@ -83,7 +96,7 @@ impl<'b> PgTypeToArrowArray<Vec<Option<CrunchyMap<'b>>>>
             _ => panic!("Expected List field"),
         };
 
-        let maps = self.into_iter().flatten().flatten().collect::<Vec<_>>();
+        let maps = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
 
         let (map_offsets, map_nulls) = arrow_map_offsets(&maps);
 
@@ -96,9 +109,9 @@ impl<'b> PgTypeToArrowArray<Vec<Option<CrunchyMap<'b>>>>
 
         for map in maps {
             if let Some(map) = map {
-                entries.push(map.entries);
+                entries.push(Some(map.entries));
             } else {
-                entries.push(vec![]);
+                entries.push(None);
             }
         }
 
@@ -106,15 +119,12 @@ impl<'b> PgTypeToArrowArray<Vec<Option<CrunchyMap<'b>>>>
 
         let entries_typoid = domain_array_base_elem_typoid(context.typoid);
 
-        let entries_tupledesc = tuple_desc(entries_typoid, -1);
-
         let entries_context = PgToArrowPerAttributeContext::new(
             context.name,
             entries_typoid,
             context.typmod,
             entries_field.clone(),
-        )
-        .with_tupledesc(entries_tupledesc);
+        );
 
         let (_, entries_array) = entries.to_arrow_array(entries_context);
 
diff --git a/src/arrow_parquet/pg_to_arrow/numeric.rs b/src/arrow_parquet/pg_to_arrow/numeric.rs
index 5772ce5..0b34a88 100644
--- a/src/arrow_parquet/pg_to_arrow/numeric.rs
+++ b/src/arrow_parquet/pg_to_arrow/numeric.rs
@@ -36,14 +36,19 @@ impl PgTypeToArrowArray<AnyNumeric> for Vec<Option<AnyNumeric>> {
 }
 
 // Numeric[]
-impl PgTypeToArrowArray<Vec<Option<AnyNumeric>>> for Vec<Option<Vec<Option<AnyNumeric>>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, AnyNumeric>> for Vec<Option<pgrx::Array<'_, AnyNumeric>>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        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 precision = extract_precision_from_numeric_typmod(context.typmod);
         let scale = extract_scale_from_numeric_typmod(context.typmod);
 
-        let numerics = self
+        let numerics = pg_array
             .into_iter()
             .flatten()
             .flatten()
diff --git a/src/arrow_parquet/pg_to_arrow/oid.rs b/src/arrow_parquet/pg_to_arrow/oid.rs
index 7651bf0..775cfbb 100644
--- a/src/arrow_parquet/pg_to_arrow/oid.rs
+++ b/src/arrow_parquet/pg_to_arrow/oid.rs
@@ -26,11 +26,16 @@ impl PgTypeToArrowArray<Oid> for Vec<Option<Oid>> {
 }
 
 // Oid[]
-impl PgTypeToArrowArray<Vec<Option<Oid>>> for Vec<Option<Vec<Option<Oid>>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, Oid>> for Vec<Option<pgrx::Array<'_, Oid>>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        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 = self.into_iter().flatten().flatten().collect::<Vec<_>>();
+        let oids = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
         let oids = oids
             .into_iter()
             .map(|x| x.map(|x| x.as_u32()))
diff --git a/src/arrow_parquet/pg_to_arrow/record.rs b/src/arrow_parquet/pg_to_arrow/record.rs
index fc3da23..bb15ae1 100644
--- a/src/arrow_parquet/pg_to_arrow/record.rs
+++ b/src/arrow_parquet/pg_to_arrow/record.rs
@@ -8,7 +8,7 @@ use pgrx::{heap_tuple::PgHeapTuple, AllocatedByRust};
 
 use crate::{
     arrow_parquet::{arrow_utils::arrow_array_offsets, pg_to_arrow::PgTypeToArrowArray},
-    pgrx_utils::collect_valid_attributes,
+    pgrx_utils::{collect_valid_attributes, tuple_desc},
 };
 
 use super::{collect_attribute_array_from_tuples, PgToArrowPerAttributeContext};
@@ -25,15 +25,15 @@ impl PgTypeToArrowArray<PgHeapTuple<'_, AllocatedByRust>>
             _ => panic!("Expected Struct field"),
         };
 
-        let tupledesc = context.tupledesc.expect("Expected tuple descriptor");
+        let tupledesc = tuple_desc(context.typoid, context.typmod);
 
         let include_generated_columns = true;
         let attributes = collect_valid_attributes(&tupledesc, include_generated_columns);
 
-        let mut tuples = self;
-
         let mut struct_attribute_arrays = vec![];
 
+        let tuples = self;
+
         for attribute in attributes {
             let attribute_name = attribute.name();
             let attribute_typoid = attribute.type_oid().value();
@@ -44,16 +44,14 @@ impl PgTypeToArrowArray<PgHeapTuple<'_, AllocatedByRust>>
                 .find(|field| field.name() == attribute_name)
                 .expect("Expected attribute field");
 
-            let (_, array, tups) = collect_attribute_array_from_tuples(
-                tuples,
-                tupledesc.clone(),
+            let (_, array) = collect_attribute_array_from_tuples(
+                &tuples,
                 attribute_name,
                 attribute_typoid,
                 attribute_typmod,
                 attribute_field.clone(),
             );
 
-            tuples = tups;
             struct_attribute_arrays.push(array);
         }
 
@@ -70,11 +68,19 @@ impl PgTypeToArrowArray<PgHeapTuple<'_, AllocatedByRust>>
 }
 
 // PgHeapTuple[]
-impl PgTypeToArrowArray<Vec<Option<PgHeapTuple<'_, AllocatedByRust>>>>
-    for Vec<Option<Vec<Option<PgHeapTuple<'_, AllocatedByRust>>>>>
+impl PgTypeToArrowArray<pgrx::Array<'_, PgHeapTuple<'_, AllocatedByRust>>>
+    for Vec<Option<pgrx::Array<'_, PgHeapTuple<'_, AllocatedByRust>>>>
 {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        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 list_field = context.field;
 
@@ -83,15 +89,14 @@ impl PgTypeToArrowArray<Vec<Option<PgHeapTuple<'_, AllocatedByRust>>>>
             _ => panic!("Expected List field"),
         };
 
-        let tuples = self.into_iter().flatten().flatten().collect::<Vec<_>>();
+        let tuples = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
 
         let tuples_context = PgToArrowPerAttributeContext::new(
             context.name,
             context.typoid,
             context.typmod,
             struct_field.clone(),
-        )
-        .with_tupledesc(context.tupledesc.expect("Expected tuple descriptor"));
+        );
 
         let (struct_field, struct_array) = tuples.to_arrow_array(tuples_context);
 
diff --git a/src/arrow_parquet/pg_to_arrow/text.rs b/src/arrow_parquet/pg_to_arrow/text.rs
index 3a79987..a5e9035 100644
--- a/src/arrow_parquet/pg_to_arrow/text.rs
+++ b/src/arrow_parquet/pg_to_arrow/text.rs
@@ -19,11 +19,16 @@ impl PgTypeToArrowArray<String> for Vec<Option<String>> {
 }
 
 // Text[]
-impl PgTypeToArrowArray<Vec<Option<String>>> for Vec<Option<Vec<Option<String>>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, String>> for Vec<Option<pgrx::Array<'_, String>>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        let pg_array = self
+            .into_iter()
+            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
+            .collect::<Vec<_>>();
 
-        let texts = self.into_iter().flatten().flatten().collect::<Vec<_>>();
+        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+
+        let texts = pg_array.into_iter().flatten().flatten().collect::<Vec<_>>();
 
         let text_array = StringArray::from(texts);
 
diff --git a/src/arrow_parquet/pg_to_arrow/time.rs b/src/arrow_parquet/pg_to_arrow/time.rs
index 1774c85..6c31df8 100644
--- a/src/arrow_parquet/pg_to_arrow/time.rs
+++ b/src/arrow_parquet/pg_to_arrow/time.rs
@@ -5,7 +5,7 @@ use arrow::{
     datatypes::FieldRef,
 };
 use arrow_schema::DataType;
-use pgrx::Time;
+use pgrx::datum::Time;
 
 use crate::{
     arrow_parquet::{arrow_utils::arrow_array_offsets, pg_to_arrow::PgTypeToArrowArray},
@@ -29,11 +29,16 @@ impl PgTypeToArrowArray<Time> for Vec<Option<Time>> {
 }
 
 // Time[]
-impl PgTypeToArrowArray<Vec<Option<Time>>> for Vec<Option<Vec<Option<Time>>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, Time>> for Vec<Option<pgrx::Array<'_, Time>>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        let pg_array = self
+            .into_iter()
+            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
+            .collect::<Vec<_>>();
 
-        let times = self
+        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+
+        let times = pg_array
             .into_iter()
             .flatten()
             .flatten()
diff --git a/src/arrow_parquet/pg_to_arrow/timestamp.rs b/src/arrow_parquet/pg_to_arrow/timestamp.rs
index 1ee08e3..cfadf31 100644
--- a/src/arrow_parquet/pg_to_arrow/timestamp.rs
+++ b/src/arrow_parquet/pg_to_arrow/timestamp.rs
@@ -5,7 +5,7 @@ use arrow::{
     datatypes::FieldRef,
 };
 use arrow_schema::DataType;
-use pgrx::Timestamp;
+use pgrx::datum::Timestamp;
 
 use crate::{
     arrow_parquet::{arrow_utils::arrow_array_offsets, pg_to_arrow::PgTypeToArrowArray},
@@ -29,11 +29,16 @@ impl PgTypeToArrowArray<Timestamp> for Vec<Option<Timestamp>> {
 }
 
 // Timestamp[]
-impl PgTypeToArrowArray<Vec<Option<Timestamp>>> for Vec<Option<Vec<Option<Timestamp>>>> {
+impl PgTypeToArrowArray<pgrx::Array<'_, Timestamp>> for Vec<Option<pgrx::Array<'_, Timestamp>>> {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        let pg_array = self
+            .into_iter()
+            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
+            .collect::<Vec<_>>();
 
-        let timestamps = self
+        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+
+        let timestamps = pg_array
             .into_iter()
             .flatten()
             .flatten()
diff --git a/src/arrow_parquet/pg_to_arrow/timestamptz.rs b/src/arrow_parquet/pg_to_arrow/timestamptz.rs
index 5ce6cdb..59d067f 100644
--- a/src/arrow_parquet/pg_to_arrow/timestamptz.rs
+++ b/src/arrow_parquet/pg_to_arrow/timestamptz.rs
@@ -5,7 +5,7 @@ use arrow::{
     datatypes::FieldRef,
 };
 use arrow_schema::DataType;
-use pgrx::TimestampWithTimeZone;
+use pgrx::datum::TimestampWithTimeZone;
 
 use crate::{
     arrow_parquet::{arrow_utils::arrow_array_offsets, pg_to_arrow::PgTypeToArrowArray},
@@ -29,13 +29,18 @@ impl PgTypeToArrowArray<TimestampWithTimeZone> for Vec<Option<TimestampWithTimeZ
 }
 
 // TimestampTz[]
-impl PgTypeToArrowArray<Vec<Option<TimestampWithTimeZone>>>
-    for Vec<Option<Vec<Option<TimestampWithTimeZone>>>>
+impl PgTypeToArrowArray<pgrx::Array<'_, TimestampWithTimeZone>>
+    for Vec<Option<pgrx::Array<'_, TimestampWithTimeZone>>>
 {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        let pg_array = self
+            .into_iter()
+            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
+            .collect::<Vec<_>>();
 
-        let timestamptzs = self
+        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+
+        let timestamptzs = pg_array
             .into_iter()
             .flatten()
             .flatten()
diff --git a/src/arrow_parquet/pg_to_arrow/timetz.rs b/src/arrow_parquet/pg_to_arrow/timetz.rs
index 7eb2eac..4fe2e55 100644
--- a/src/arrow_parquet/pg_to_arrow/timetz.rs
+++ b/src/arrow_parquet/pg_to_arrow/timetz.rs
@@ -5,7 +5,7 @@ use arrow::{
     datatypes::FieldRef,
 };
 use arrow_schema::DataType;
-use pgrx::TimeWithTimeZone;
+use pgrx::datum::TimeWithTimeZone;
 
 use crate::{
     arrow_parquet::{arrow_utils::arrow_array_offsets, pg_to_arrow::PgTypeToArrowArray},
@@ -29,13 +29,18 @@ impl PgTypeToArrowArray<TimeWithTimeZone> for Vec<Option<TimeWithTimeZone>> {
 }
 
 // TimeTz[]
-impl PgTypeToArrowArray<Vec<Option<TimeWithTimeZone>>>
-    for Vec<Option<Vec<Option<TimeWithTimeZone>>>>
+impl PgTypeToArrowArray<pgrx::Array<'_, TimeWithTimeZone>>
+    for Vec<Option<pgrx::Array<'_, TimeWithTimeZone>>>
 {
     fn to_arrow_array(self, context: PgToArrowPerAttributeContext) -> (FieldRef, ArrayRef) {
-        let (offsets, nulls) = arrow_array_offsets(&self);
+        let pg_array = self
+            .into_iter()
+            .map(|v| v.map(|pg_array| pg_array.iter().collect::<Vec<_>>()))
+            .collect::<Vec<_>>();
 
-        let timetzs = self
+        let (offsets, nulls) = arrow_array_offsets(&pg_array);
+
+        let timetzs = pg_array
             .into_iter()
             .flatten()
             .flatten()
diff --git a/src/bin/pgrx_embed.rs b/src/bin/pgrx_embed.rs
new file mode 100644
index 0000000..bf56530
--- /dev/null
+++ b/src/bin/pgrx_embed.rs
@@ -0,0 +1,2 @@
+// magic is required due to PR https://github.com/pgcentralfoundation/pgrx/pull/1468
+::pgrx::pgrx_embed!();
diff --git a/src/lib.rs b/src/lib.rs
index e81c44c..eca35f8 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,9 +1,10 @@
-use parquet_copy_hook::hook::PARQUET_COPY_HOOK;
-use pgrx::{prelude::*, register_hook};
+use parquet_copy_hook::hook::init_parquet_copy_hook;
+use pgrx::prelude::*;
 
 mod arrow_parquet;
 mod parquet_copy_hook;
 mod parquet_udfs;
+mod pgrx_missing_declerations;
 mod pgrx_utils;
 mod type_compat;
 
@@ -18,7 +19,7 @@ pgrx::pg_module_magic!();
 #[allow(static_mut_refs)]
 #[pg_guard]
 pub extern "C" fn _PG_init() {
-    unsafe { register_hook(&mut PARQUET_COPY_HOOK) };
+    init_parquet_copy_hook();
 }
 
 #[cfg(any(test, feature = "pg_test"))]
@@ -33,8 +34,9 @@ mod tests {
     use crate::type_compat::geometry::Geometry;
     use pgrx::pg_sys::Oid;
     use pgrx::{
-        composite_type, pg_test, AnyNumeric, Date, FromDatum, IntoDatum, Spi, Time,
-        TimeWithTimeZone, Timestamp, TimestampWithTimeZone,
+        composite_type,
+        datum::{Date, Time, TimeWithTimeZone, Timestamp, TimestampWithTimeZone},
+        pg_test, AnyNumeric, FromDatum, IntoDatum, Spi,
     };
     enum CopyOptionValue {
         StringOption(String),
@@ -526,11 +528,28 @@ mod tests {
 
     #[pg_test]
     fn test_bytea_array() {
-        let test_table = TestTable::<Vec<Option<Vec<u8>>>>::new("bytea[]".into());
+        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);",
         );
-        test_helper(test_table);
+        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());
+            }
+
+            if expected.is_some() {
+                assert!(actual.is_some());
+
+                let expected = expected.unwrap();
+                let actual = actual.unwrap();
+
+                for (expected, actual) in expected.iter().zip(actual.iter()) {
+                    assert_eq!(expected, actual);
+                }
+            }
+        }
     }
 
     #[pg_test]
@@ -694,18 +713,7 @@ mod tests {
         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());
-            }
-
-            if expected.is_some() {
-                assert!(actual.is_some());
-
-                let expected = expected.unwrap();
-                let actual = actual.unwrap();
-
-                assert_eq!(expected.0, actual.0);
-            }
+            assert_eq!(expected, actual);
         }
     }
 
@@ -717,31 +725,7 @@ mod tests {
         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());
-            }
-
-            if expected.is_some() {
-                assert!(actual.is_some());
-
-                let expected = expected.unwrap();
-                let actual = actual.unwrap();
-
-                for (expected, actual) in expected.into_iter().zip(actual.into_iter()) {
-                    if expected.is_none() {
-                        assert!(actual.is_none());
-                    }
-
-                    if expected.is_some() {
-                        assert!(actual.is_some());
-
-                        let expected = expected.unwrap();
-                        let actual = actual.unwrap();
-
-                        assert_eq!(expected.0, actual.0);
-                    }
-                }
-            }
+            assert_eq!(expected, actual);
         }
     }
 
@@ -753,18 +737,7 @@ mod tests {
         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());
-            }
-
-            if expected.is_some() {
-                assert!(actual.is_some());
-
-                let expected = expected.unwrap();
-                let actual = actual.unwrap();
-
-                assert_eq!(expected.0, actual.0);
-            }
+            assert_eq!(expected, actual);
         }
     }
 
@@ -775,31 +748,7 @@ mod tests {
         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());
-            }
-
-            if expected.is_some() {
-                assert!(actual.is_some());
-
-                let expected = expected.unwrap();
-                let actual = actual.unwrap();
-
-                for (expected, actual) in expected.into_iter().zip(actual.into_iter()) {
-                    if expected.is_none() {
-                        assert!(actual.is_none());
-                    }
-
-                    if expected.is_some() {
-                        assert!(actual.is_some());
-
-                        let expected = expected.unwrap();
-                        let actual = actual.unwrap();
-
-                        assert_eq!(expected.0, actual.0);
-                    }
-                }
-            }
+            assert_eq!(expected, actual);
         }
     }
 
@@ -934,10 +883,10 @@ mod tests {
                 );
 
                 let expected_dogs = expected
-                    .get_by_name::<Vec<Option<composite_type!("dog")>>>("dogs")
+                    .get_by_name::<pgrx::Array<composite_type!("dog")>>("dogs")
                     .unwrap();
                 let actual_dogs = actual
-                    .get_by_name::<Vec<Option<composite_type!("dog")>>>("dogs")
+                    .get_by_name::<pgrx::Array<composite_type!("dog")>>("dogs")
                     .unwrap();
 
                 if expected_dogs.is_none() {
@@ -948,9 +897,7 @@ mod tests {
                     let expected_dogs = expected_dogs.unwrap();
                     let actual_dogs = actual_dogs.unwrap();
 
-                    for (expected_dog, actual_dog) in
-                        expected_dogs.into_iter().zip(actual_dogs.into_iter())
-                    {
+                    for (expected_dog, actual_dog) in expected_dogs.iter().zip(actual_dogs.iter()) {
                         if expected_dog.is_none() {
                             assert!(actual_dog.is_none());
                         } else if expected_dog.is_some() {
@@ -973,14 +920,28 @@ mod tests {
                 }
 
                 let expected_lucky_numbers = expected
-                    .get_by_name::<Vec<Option<i32>>>("lucky_numbers")
+                    .get_by_name::<pgrx::Array<i32>>("lucky_numbers")
                     .unwrap();
 
                 let actual_lucky_numbers = actual
-                    .get_by_name::<Vec<Option<i32>>>("lucky_numbers")
+                    .get_by_name::<pgrx::Array<i32>>("lucky_numbers")
                     .unwrap();
 
-                assert_eq!(expected_lucky_numbers, actual_lucky_numbers);
+                if expected_lucky_numbers.is_none() {
+                    assert!(actual_lucky_numbers.is_none());
+                } else if expected_lucky_numbers.is_some() {
+                    assert!(actual_lucky_numbers.is_some());
+
+                    let expected_lucky_numbers = expected_lucky_numbers.unwrap();
+                    let actual_lucky_numbers = actual_lucky_numbers.unwrap();
+
+                    for (expected_lucky_number, actual_lucky_number) in expected_lucky_numbers
+                        .into_iter()
+                        .zip(actual_lucky_numbers.into_iter())
+                    {
+                        assert_eq!(expected_lucky_number, actual_lucky_number);
+                    }
+                }
             }
         }
 
@@ -1076,6 +1037,18 @@ 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/parquet_copy_hook/copy_from.rs b/src/parquet_copy_hook/copy_from.rs
index 151f525..34868cd 100644
--- a/src/parquet_copy_hook/copy_from.rs
+++ b/src/parquet_copy_hook/copy_from.rs
@@ -2,9 +2,9 @@ use pgrx::{
     ereport, pg_guard,
     pg_sys::{
         self, canonicalize_qual, coerce_to_boolean, eval_const_expressions, make_ands_explicit,
-        make_parsestate, AccessShareLock, AsPgCStr, CopyStmt, CreateTemplateTupleDesc, List,
-        ParseExprKind_EXPR_KIND_COPY_WHERE, ParseNamespaceItem, ParseState, RowExclusiveLock,
-        TupleDescInitEntry,
+        make_parsestate, transformExpr, AccessShareLock, AsPgCStr, BeginCopyFrom, CopyFrom,
+        CopyStmt, CreateTemplateTupleDesc, EndCopyFrom, List, ParseExprKind, ParseNamespaceItem,
+        ParseState, RowExclusiveLock, TupleDescInitEntry,
     },
     void_mut_ptr, PgBox, PgList, PgLogLevel, PgRelation, PgSqlErrorCode, PgTupleDesc,
 };
@@ -12,10 +12,11 @@ use pgrx::{
 use crate::{
     arrow_parquet::parquet_reader::ParquetReaderContext,
     parquet_copy_hook::copy_utils::{
-        addNSItemToQuery, addRangeTableEntryForRelation, add_binary_format_option,
-        assign_expr_collations, copy_lock_mode, copy_relation_oid, copy_stmt_filename,
-        is_copy_from_parquet_stmt, transformExpr, BeginCopyFrom, CopyFrom, CopyFromState,
-        EndCopyFrom,
+        add_binary_format_option, copy_lock_mode, copy_relation_oid, copy_stmt_filename,
+        is_copy_from_parquet_stmt,
+    },
+    pgrx_missing_declerations::{
+        addNSItemToQuery, addRangeTableEntryForRelation, assign_expr_collations,
     },
 };
 
@@ -40,9 +41,9 @@ pub(crate) fn pop_parquet_reader_context(throw_error: bool) {
             PgSqlErrorCode::ERRCODE_INTERNAL_ERROR,
             "parquet reader context stack is already empty"
         );
+    } else {
+        current_parquet_reader_context.take();
     }
-
-    current_parquet_reader_context.take();
 }
 
 pub(crate) fn push_parquet_reader_context(reader_ctx: ParquetReaderContext) {
@@ -83,7 +84,7 @@ pub(crate) fn execute_copy_from(
     pstmt: PgBox<pg_sys::PlannedStmt>,
     query_string: &core::ffi::CStr,
     query_env: PgBox<pg_sys::QueryEnvironment>,
-) -> i64 {
+) -> u64 {
     let rel_oid = copy_relation_oid(&pstmt);
     let lock_mode = copy_lock_mode(&pstmt);
     let relation = unsafe { PgRelation::with_lock(rel_oid, lock_mode) };
@@ -115,7 +116,7 @@ pub(crate) fn execute_copy_from(
 
         let copy_options = add_binary_format_option(&pstmt);
 
-        let copy_from_state: CopyFromState = BeginCopyFrom(
+        let copy_from_state = BeginCopyFrom(
             pstate.as_ptr(),
             relation.as_ptr(),
             where_clause.as_ptr(),
@@ -153,7 +154,7 @@ fn copy_from_transform_where_clause(
         transformExpr(
             pstate.as_ptr(),
             where_clause.as_ptr(),
-            ParseExprKind_EXPR_KIND_COPY_WHERE,
+            ParseExprKind::EXPR_KIND_COPY_WHERE,
         )
     };
 
diff --git a/src/parquet_copy_hook/copy_to.rs b/src/parquet_copy_hook/copy_to.rs
index 37bf595..f92356a 100644
--- a/src/parquet_copy_hook/copy_to.rs
+++ b/src/parquet_copy_hook/copy_to.rs
@@ -1,17 +1,15 @@
 use pgrx::{
     is_a,
     pg_sys::{
-        self, makeRangeVar, pg_analyze_and_rewrite_fixedparams, pg_plan_query,
-        CommandTag_CMDTAG_COPY, CopyStmt, CreateNewPortal, DestReceiver, GetActiveSnapshot,
-        NodeTag::T_CopyStmt, PlannedStmt, PortalDefineQuery, PortalDrop, QueryCompletion, RawStmt,
-        CURSOR_OPT_PARALLEL_OK,
+        self, makeRangeVar, pg_analyze_and_rewrite_fixedparams, pg_plan_query, CommandTag,
+        CopyStmt, CreateNewPortal, DestReceiver, GetActiveSnapshot, NodeTag::T_CopyStmt,
+        PlannedStmt, PortalDefineQuery, PortalDrop, PortalRun, PortalStart, QueryCompletion,
+        RawStmt, CURSOR_OPT_PARALLEL_OK,
     },
     AllocatedByRust, PgBox, PgList, PgRelation,
 };
 
-use crate::parquet_copy_hook::copy_utils::{
-    copy_has_relation, copy_lock_mode, copy_relation_oid, PortalRun, PortalStart,
-};
+use crate::parquet_copy_hook::copy_utils::{copy_has_relation, copy_lock_mode, copy_relation_oid};
 
 pub(crate) fn execute_copy_to_with_dest_receiver(
     pstmt: &PgBox<pg_sys::PlannedStmt>,
@@ -66,7 +64,7 @@ pub(crate) fn execute_copy_to_with_dest_receiver(
             portal.as_ptr(),
             std::ptr::null(),
             query_string.as_ptr(),
-            CommandTag_CMDTAG_COPY,
+            CommandTag::CMDTAG_COPY,
             plans.as_ptr(),
             std::ptr::null_mut(),
         );
@@ -76,7 +74,7 @@ pub(crate) fn execute_copy_to_with_dest_receiver(
 
         // run portal
         let mut completion_tag = QueryCompletion {
-            commandTag: CommandTag_CMDTAG_COPY,
+            commandTag: CommandTag::CMDTAG_COPY,
             nprocessed: 0,
         };
 
diff --git a/src/parquet_copy_hook/copy_to_dest_receiver.rs b/src/parquet_copy_hook/copy_to_dest_receiver.rs
index 625bf87..f9fa23c 100644
--- a/src/parquet_copy_hook/copy_to_dest_receiver.rs
+++ b/src/parquet_copy_hook/copy_to_dest_receiver.rs
@@ -1,5 +1,5 @@
 use pg_sys::{
-    AsPgCStr, BlessTupleDesc, CommandDest_DestCopyOut, CurrentMemoryContext, Datum, DestReceiver,
+    AsPgCStr, BlessTupleDesc, CommandDest, CurrentMemoryContext, Datum, DestReceiver,
     HeapTupleData, List, MemoryContext, TupleDesc, TupleTableSlot,
 };
 use pgrx::{prelude::*, PgList, PgMemoryContexts, PgTupleDesc};
@@ -9,7 +9,7 @@ use crate::{
         codec::ParquetCodecOption, parquet_writer::ParquetWriterContext,
         schema_visitor::parquet_schema_string_from_tupledesc,
     },
-    parquet_copy_hook::copy_utils::slot_getallattrs,
+    pgrx_missing_declerations::slot_getallattrs,
     pgrx_utils::collect_valid_attributes,
 };
 
@@ -35,7 +35,9 @@ pub(crate) fn peek_parquet_writer_context() -> Option<&'static mut ParquetWriter
 pub(crate) fn pop_parquet_writer_context(throw_error: bool) {
     let current_parquet_writer_context = unsafe { PARQUET_WRITER_CONTEXT_STACK.pop() };
 
-    if current_parquet_writer_context.is_none() {
+    if let Some(current_parquet_writer_context) = current_parquet_writer_context {
+        current_parquet_writer_context.close();
+    } else {
         let level = if throw_error {
             PgLogLevel::ERROR
         } else {
@@ -48,8 +50,6 @@ pub(crate) fn pop_parquet_writer_context(throw_error: bool) {
             "parquet writer context stack is already empty"
         );
     }
-
-    current_parquet_writer_context.unwrap().close();
 }
 
 pub(crate) fn push_parquet_writer_context(writer_ctx: ParquetWriterContext<'static>) {
@@ -220,7 +220,7 @@ pub extern "C" fn create_copy_to_parquet_dest_receiver(
     parquet_dest.dest.rStartup = Some(copy_startup);
     parquet_dest.dest.rShutdown = Some(copy_shutdown);
     parquet_dest.dest.rDestroy = Some(copy_destroy);
-    parquet_dest.dest.mydest = CommandDest_DestCopyOut;
+    parquet_dest.dest.mydest = CommandDest::DestCopyOut;
     parquet_dest.filename = filename;
     parquet_dest.tupledesc = std::ptr::null_mut();
     parquet_dest.natts = 0;
diff --git a/src/parquet_copy_hook/copy_utils.rs b/src/parquet_copy_hook/copy_utils.rs
index 8d9061b..fb927be 100644
--- a/src/parquet_copy_hook/copy_utils.rs
+++ b/src/parquet_copy_hook/copy_utils.rs
@@ -3,11 +3,10 @@ use std::str::FromStr;
 use pgrx::{
     is_a,
     pg_sys::{
-        self, defGetInt64, defGetString, AccessShareLock, Alias, CopyStmt, DefElem, DestReceiver,
-        NodeTag::T_CopyStmt, Oid, ParamListInfo, ParseExprKind, ParseNamespaceItem, ParseState,
-        Portal, QueryCompletion, Relation, RowExclusiveLock, Snapshot, TupleTableSlot,
+        self, defGetInt64, defGetString, AccessShareLock, CopyStmt, DefElem, NodeTag::T_CopyStmt,
+        Oid, RowExclusiveLock,
     },
-    void_mut_ptr, PgBox, PgList, PgRelation,
+    PgBox, PgList, PgRelation,
 };
 
 use crate::arrow_parquet::codec::{all_supported_codecs, FromPath, ParquetCodecOption};
@@ -276,93 +275,3 @@ pub(crate) fn copy_relation_oid(pstmt: &PgBox<pg_sys::PlannedStmt>) -> Oid {
     let relation = PgRelation::open_with_name_and_share_lock(relname).unwrap();
     relation.rd_id
 }
-
-// needed to declare these functions since they are not available in pg_sys yet
-#[repr(C)]
-pub(crate) struct CopyFromStateData {
-    _data: [u8; 0],
-    _marker: core::marker::PhantomData<(*mut u8, core::marker::PhantomPinned)>,
-}
-
-pub(crate) type CopyFromState = *mut CopyFromStateData;
-
-#[allow(improper_ctypes)]
-extern "C" {
-    pub(crate) fn PortalStart(
-        portal: Portal,
-        params: ParamListInfo,
-        eflags: ::std::os::raw::c_int,
-        snapshot: Snapshot,
-    );
-    pub(crate) fn PortalRun(
-        portal: Portal,
-        count: ::std::os::raw::c_long,
-        is_top_level: bool,
-        run_once: bool,
-        dest: *mut DestReceiver,
-        alt_dest: *mut DestReceiver,
-        completion_tag: *mut QueryCompletion,
-    );
-
-    pub(crate) fn BeginCopyFrom(
-        parse_state: *mut ParseState,
-        rel: Relation,
-        where_clause: *mut pg_sys::Node,
-        filename: *const i8,
-        is_program: bool,
-        data_source_cb: Option<extern "C" fn(void_mut_ptr, i32, i32) -> i32>,
-        attnamelist: *mut pg_sys::List,
-        options: *mut pg_sys::List,
-    ) -> CopyFromState;
-
-    pub(crate) fn CopyFrom(cstate: CopyFromState) -> i64;
-
-    pub(crate) fn EndCopyFrom(cstate: CopyFromState);
-
-    pub(crate) fn addRangeTableEntryForRelation(
-        pstate: *mut ParseState,
-        rel: Relation,
-        lockmode: i32,
-        alias: *mut Alias,
-        inh: bool,
-        inFromCl: bool,
-    ) -> *mut ParseNamespaceItem;
-
-    pub(crate) fn addNSItemToQuery(
-        pstate: *mut ParseState,
-        nsitem: *mut ParseNamespaceItem,
-        add_to_join_list: bool,
-        add_to_rel_namespace: bool,
-        add_to_var_namespace: bool,
-    );
-
-    pub(crate) fn transformExpr(
-        pstate: *mut ParseState,
-        expr: *mut pg_sys::Node,
-        expr_kind: ParseExprKind,
-    ) -> *mut pg_sys::Node;
-
-    pub(crate) fn assign_expr_collations(
-        pstate: *mut ParseState,
-        expr: *mut pg_sys::Node,
-    ) -> *mut pg_sys::Node;
-}
-
-/*
- * slot_getallattrs
- *		This function forces all the entries of the slot's Datum/isnull
- *		arrays to be valid.  The caller may then extract data directly
- *		from those arrays instead of using slot_getattr.
- */
-pub(crate) fn slot_getallattrs(slot: *mut TupleTableSlot) {
-    // copied from Postgres since this method was inlined in the original code
-    // (not found in pg_sys)
-    // handles select * from table
-    unsafe {
-        let slot = PgBox::from_pg(slot);
-        let tts_tupledesc = PgBox::from_pg(slot.tts_tupleDescriptor);
-        if (slot.tts_nvalid as i32) < tts_tupledesc.natts {
-            pg_sys::slot_getsomeattrs_int(slot.as_ptr(), tts_tupledesc.natts);
-        }
-    };
-}
diff --git a/src/parquet_copy_hook/hook.rs b/src/parquet_copy_hook/hook.rs
index dc747d8..69074a6 100644
--- a/src/parquet_copy_hook/hook.rs
+++ b/src/parquet_copy_hook/hook.rs
@@ -1,5 +1,7 @@
-use pg_sys::CommandTag_CMDTAG_COPY;
-use pgrx::{prelude::*, HookResult, PgHooks};
+use std::ffi::CStr;
+
+use pg_sys::{standard_ProcessUtility, CommandTag, ProcessUtility_hook, ProcessUtility_hook_type};
+use pgrx::prelude::*;
 
 use crate::parquet_copy_hook::{
     copy_to_dest_receiver::create_copy_to_parquet_dest_receiver,
@@ -16,104 +18,121 @@ use super::{
     copy_utils::{copy_stmt_codec, validate_copy_from_options, validate_copy_to_options},
 };
 
-pub(crate) static mut PARQUET_COPY_HOOK: ParquetCopyHook = ParquetCopyHook {};
-
-pub(crate) struct ParquetCopyHook {}
-
-impl ParquetCopyHook {}
-
-impl PgHooks for ParquetCopyHook {
-    fn process_utility_hook(
-        &mut self,
-        pstmt: PgBox<pg_sys::PlannedStmt>,
-        query_string: &core::ffi::CStr,
-        read_only_tree: Option<bool>,
-        context: pg_sys::ProcessUtilityContext,
-        params: PgBox<pg_sys::ParamListInfoData>,
-        query_env: PgBox<pg_sys::QueryEnvironment>,
-        dest: PgBox<pg_sys::DestReceiver>,
-        completion_tag: *mut pg_sys::QueryCompletion,
-        prev_hook: fn(
-            pstmt: PgBox<pg_sys::PlannedStmt>,
-            query_string: &core::ffi::CStr,
-            read_only_tree: Option<bool>,
-            context: pg_sys::ProcessUtilityContext,
-            params: PgBox<pg_sys::ParamListInfoData>,
-            query_env: PgBox<pg_sys::QueryEnvironment>,
-            dest: PgBox<pg_sys::DestReceiver>,
-            completion_tag: *mut pg_sys::QueryCompletion,
-        ) -> HookResult<()>,
-    ) -> HookResult<()> {
-        if is_copy_to_parquet_stmt(&pstmt) {
-            validate_copy_to_options(&pstmt);
-
-            let filename = copy_stmt_filename(&pstmt);
-            let row_group_size = copy_stmt_row_group_size_option(&pstmt);
-            let codec = copy_stmt_codec(&pstmt);
-
-            PgTryBuilder::new(|| {
-                let parquet_dest =
-                    create_copy_to_parquet_dest_receiver(filename, row_group_size, codec);
-
-                let parquet_dest = unsafe { PgBox::from_pg(parquet_dest) };
-
-                let nprocessed = execute_copy_to_with_dest_receiver(
-                    &pstmt,
-                    query_string,
-                    params,
-                    query_env,
-                    parquet_dest,
-                );
-
-                let mut completion_tag = unsafe { PgBox::from_pg(completion_tag) };
-                if !completion_tag.is_null() {
-                    completion_tag.nprocessed = nprocessed;
-                    completion_tag.commandTag = CommandTag_CMDTAG_COPY;
-                }
-            })
-            .catch_others(|cause| {
-                // make sure to pop the parquet writer context
-                let throw_error = false;
-                pop_parquet_writer_context(throw_error);
-
-                cause.rethrow()
-            })
-            .execute();
-
-            return HookResult::new(());
-        } else if is_copy_from_parquet_stmt(&pstmt) {
-            validate_copy_from_options(&pstmt);
-
-            PgTryBuilder::new(|| {
-                let nprocessed = execute_copy_from(pstmt, query_string, query_env);
-
-                let mut completion_tag = unsafe { PgBox::from_pg(completion_tag) };
-                if !completion_tag.is_null() {
-                    completion_tag.nprocessed = nprocessed as _;
-                    completion_tag.commandTag = CommandTag_CMDTAG_COPY;
-                }
-            })
-            .catch_others(|cause| {
-                // make sure to pop the parquet reader context
-                let throw_error = false;
-                pop_parquet_reader_context(throw_error);
-
-                cause.rethrow()
-            })
-            .execute();
-
-            return HookResult::new(());
+static mut PREV_PROCESS_UTILITY_HOOK: ProcessUtility_hook_type = None;
+
+#[pg_guard]
+#[no_mangle]
+pub(crate) extern "C" fn init_parquet_copy_hook() {
+    unsafe {
+        if ProcessUtility_hook.is_some() {
+            PREV_PROCESS_UTILITY_HOOK = ProcessUtility_hook
         }
 
-        prev_hook(
-            pstmt,
-            query_string,
-            read_only_tree,
-            context,
-            params,
-            query_env,
-            dest,
-            completion_tag,
-        )
+        ProcessUtility_hook = Some(parquet_copy_hook);
+    }
+}
+
+#[pg_guard]
+#[allow(clippy::too_many_arguments)]
+extern "C" fn parquet_copy_hook(
+    pstmt: *mut pg_sys::PlannedStmt,
+    query_string: *const i8,
+    read_only_tree: bool,
+    context: u32,
+    params: *mut pg_sys::ParamListInfoData,
+    query_env: *mut pg_sys::QueryEnvironment,
+    dest: *mut pg_sys::DestReceiver,
+    completion_tag: *mut pg_sys::QueryCompletion,
+) {
+    let pstmt = unsafe { PgBox::from_pg(pstmt) };
+    let query_string = unsafe { CStr::from_ptr(query_string) };
+    let params = unsafe { PgBox::from_pg(params) };
+    let query_env = unsafe { PgBox::from_pg(query_env) };
+
+    if is_copy_to_parquet_stmt(&pstmt) {
+        validate_copy_to_options(&pstmt);
+
+        let filename = copy_stmt_filename(&pstmt);
+        let row_group_size = copy_stmt_row_group_size_option(&pstmt);
+        let codec = copy_stmt_codec(&pstmt);
+
+        PgTryBuilder::new(|| {
+            let parquet_dest =
+                create_copy_to_parquet_dest_receiver(filename, row_group_size, codec);
+
+            let parquet_dest = unsafe { PgBox::from_pg(parquet_dest) };
+
+            let nprocessed = execute_copy_to_with_dest_receiver(
+                &pstmt,
+                query_string,
+                params,
+                query_env,
+                parquet_dest,
+            );
+
+            let mut completion_tag = unsafe { PgBox::from_pg(completion_tag) };
+            if !completion_tag.is_null() {
+                completion_tag.nprocessed = nprocessed;
+                completion_tag.commandTag = CommandTag::CMDTAG_COPY;
+            }
+        })
+        .catch_others(|cause| {
+            // make sure to pop the parquet writer context
+            let throw_error = false;
+            pop_parquet_writer_context(throw_error);
+
+            cause.rethrow()
+        })
+        .execute();
+
+        return;
+    } else if is_copy_from_parquet_stmt(&pstmt) {
+        validate_copy_from_options(&pstmt);
+
+        PgTryBuilder::new(|| {
+            let nprocessed = execute_copy_from(pstmt, query_string, query_env);
+
+            let mut completion_tag = unsafe { PgBox::from_pg(completion_tag) };
+            if !completion_tag.is_null() {
+                completion_tag.nprocessed = nprocessed;
+                completion_tag.commandTag = CommandTag::CMDTAG_COPY;
+            }
+        })
+        .catch_others(|cause| {
+            // make sure to pop the parquet reader context
+            let throw_error = false;
+            pop_parquet_reader_context(throw_error);
+
+            cause.rethrow()
+        })
+        .execute();
+
+        return;
+    }
+
+    unsafe {
+        if let Some(prev_hook) = PREV_PROCESS_UTILITY_HOOK {
+            prev_hook(
+                pstmt.into_pg(),
+                query_string.as_ptr(),
+                read_only_tree,
+                context,
+                params.into_pg(),
+                query_env.into_pg(),
+                dest,
+                completion_tag,
+            )
+        } else {
+            standard_ProcessUtility(
+                pstmt.into_pg(),
+                query_string.as_ptr(),
+                read_only_tree,
+                context,
+                params.into_pg(),
+                query_env.into_pg(),
+                dest,
+                completion_tag,
+            )
+        }
     }
 }
diff --git a/src/pgrx_missing_declerations.rs b/src/pgrx_missing_declerations.rs
new file mode 100644
index 0000000..0138236
--- /dev/null
+++ b/src/pgrx_missing_declerations.rs
@@ -0,0 +1,52 @@
+use pgrx::{
+    pg_sys::{self, Alias, Oid, ParseNamespaceItem, ParseState, Relation, TupleTableSlot},
+    PgBox,
+};
+
+#[allow(improper_ctypes)]
+extern "C" {
+    pub(crate) fn get_extension_oid(name: *const i8, missing_ok: bool) -> Oid;
+
+    pub(crate) fn get_extension_schema(ext_oid: Oid) -> Oid;
+
+    pub(crate) fn addRangeTableEntryForRelation(
+        pstate: *mut ParseState,
+        rel: Relation,
+        lockmode: i32,
+        alias: *mut Alias,
+        inh: bool,
+        inFromCl: bool,
+    ) -> *mut ParseNamespaceItem;
+
+    pub(crate) fn addNSItemToQuery(
+        pstate: *mut ParseState,
+        nsitem: *mut ParseNamespaceItem,
+        add_to_join_list: bool,
+        add_to_rel_namespace: bool,
+        add_to_var_namespace: bool,
+    );
+
+    pub(crate) fn assign_expr_collations(
+        pstate: *mut ParseState,
+        expr: *mut pg_sys::Node,
+    ) -> *mut pg_sys::Node;
+}
+
+/*
+ * slot_getallattrs
+ *		This function forces all the entries of the slot's Datum/isnull
+ *		arrays to be valid.  The caller may then extract data directly
+ *		from those arrays instead of using slot_getattr.
+ */
+pub(crate) fn slot_getallattrs(slot: *mut TupleTableSlot) {
+    // copied from Postgres since this method was inlined in the original code
+    // (not found in pg_sys)
+    // handles select * from table
+    unsafe {
+        let slot = PgBox::from_pg(slot);
+        let tts_tupledesc = PgBox::from_pg(slot.tts_tupleDescriptor);
+        if (slot.tts_nvalid as i32) < tts_tupledesc.natts {
+            pg_sys::slot_getsomeattrs_int(slot.as_ptr(), tts_tupledesc.natts);
+        }
+    };
+}
diff --git a/src/type_compat/fallback_to_text.rs b/src/type_compat/fallback_to_text.rs
index 967d9f5..dfe3bd2 100644
--- a/src/type_compat/fallback_to_text.rs
+++ b/src/type_compat/fallback_to_text.rs
@@ -2,6 +2,7 @@ use std::borrow::BorrowMut;
 
 use once_cell::sync::OnceCell;
 use pgrx::{
+    datum::UnboxDatum,
     pg_sys::{
         self, fmgr_info, getTypeInputInfo, getTypeOutputInfo, AsPgCStr, FmgrInfo,
         InputFunctionCall, InvalidOid, Oid, OutputFunctionCall,
@@ -134,3 +135,24 @@ impl FromDatum for FallbackToText {
         }
     }
 }
+
+unsafe impl UnboxDatum for FallbackToText {
+    type As<'src> = FallbackToText;
+
+    unsafe fn unbox<'src>(datum: pgrx::datum::Datum<'src>) -> Self::As<'src>
+    where
+        Self: 'src,
+    {
+        let att_cstr = OutputFunctionCall(
+            get_fallback_to_text_context().output_func.borrow_mut(),
+            datum.sans_lifetime(),
+        );
+
+        let att_val = std::ffi::CStr::from_ptr(att_cstr)
+            .to_str()
+            .unwrap()
+            .to_owned();
+
+        Self(att_val)
+    }
+}
diff --git a/src/type_compat/geometry.rs b/src/type_compat/geometry.rs
index 308a17a..e00e474 100644
--- a/src/type_compat/geometry.rs
+++ b/src/type_compat/geometry.rs
@@ -2,18 +2,15 @@ use std::{ffi::CString, ops::Deref};
 
 use once_cell::sync::OnceCell;
 use pgrx::{
+    datum::UnboxDatum,
     pg_sys::{
         self, makeString, Anum_pg_type_oid, AsPgCStr, GetSysCacheOid, InvalidOid, LookupFuncName,
-        Oid, OidFunctionCall1Coll, SysCacheIdentifier_TYPENAMENSP, BYTEAOID,
+        Oid, OidFunctionCall1Coll, SysCacheIdentifier::TYPENAMENSP, BYTEAOID,
     },
     FromDatum, IntoDatum, PgList,
 };
 
-#[allow(improper_ctypes)]
-extern "C" {
-    fn get_extension_oid(name: *const i8, missing_ok: bool) -> Oid;
-    fn get_extension_schema(ext_oid: Oid) -> Oid;
-}
+use crate::pgrx_missing_declerations::{get_extension_oid, get_extension_schema};
 
 // we need to reset the postgis context at each copy start
 static mut POSTGIS_CONTEXT: OnceCell<PostgisContext> = OnceCell::new();
@@ -116,7 +113,7 @@ impl PostgisContext {
 
         let postgis_geometry_typoid = unsafe {
             GetSysCacheOid(
-                SysCacheIdentifier_TYPENAMENSP as _,
+                TYPENAMENSP as _,
                 Anum_pg_type_oid as _,
                 postgis_geometry_type_name.into_datum().unwrap(),
                 postgis_ext_schema_oid.into_datum().unwrap(),
@@ -194,3 +191,23 @@ impl FromDatum for Geometry {
         }
     }
 }
+
+unsafe impl UnboxDatum for Geometry {
+    type As<'src> = Geometry;
+
+    unsafe fn unbox<'src>(datum: pgrx::datum::Datum<'src>) -> Self::As<'src>
+    where
+        Self: 'src,
+    {
+        let st_asbinary_func_oid = get_postgis_context()
+            .st_asbinary_funcoid
+            .expect("st_asbinary_funcoid");
+
+        let geom_datum = datum.sans_lifetime();
+
+        let wkb_datum = OidFunctionCall1Coll(st_asbinary_func_oid, InvalidOid, geom_datum);
+
+        let wkb = Vec::<u8>::from_datum(wkb_datum, false).unwrap();
+        Geometry(wkb)
+    }
+}
diff --git a/src/type_compat/map.rs b/src/type_compat/map.rs
index 2e9772f..4dd720e 100644
--- a/src/type_compat/map.rs
+++ b/src/type_compat/map.rs
@@ -1,20 +1,18 @@
 use once_cell::sync::OnceCell;
 use pgrx::{
+    datum::UnboxDatum,
     pg_sys::{
         self, Anum_pg_type_oid, AsPgCStr, GetSysCacheOid, InvalidOid, Oid,
-        SysCacheIdentifier_TYPEOID,
+        SysCacheIdentifier::TYPEOID,
     },
     prelude::PgHeapTuple,
     AllocatedByRust, FromDatum, IntoDatum,
 };
 
-use crate::pgrx_utils::is_domain_of_array_type;
-
-#[allow(improper_ctypes)]
-extern "C" {
-    fn get_extension_oid(name: *const i8, missing_ok: bool) -> Oid;
-    fn get_extension_schema(ext_oid: Oid) -> Oid;
-}
+use crate::{
+    pgrx_missing_declerations::{get_extension_oid, get_extension_schema},
+    pgrx_utils::is_domain_of_array_type,
+};
 
 // we need to reset the crunchy_map context at each copy start
 static mut CRUNCHY_MAP_CONTEXT: OnceCell<CrunchyMapContext> = OnceCell::new();
@@ -53,7 +51,7 @@ pub(crate) fn is_crunchy_map_typoid(typoid: Oid) -> bool {
 
     let found_typoid = unsafe {
         GetSysCacheOid(
-            SysCacheIdentifier_TYPEOID as _,
+            TYPEOID as _,
             Anum_pg_type_oid as _,
             typoid.into_datum().unwrap(),
             crunchy_map_ext_schema_oid.into_datum().unwrap(),
@@ -115,7 +113,7 @@ impl CrunchyMapContext {
 
 // crunchy_map is a domain type over array of key-value pairs
 pub(crate) struct CrunchyMap<'a> {
-    pub(crate) entries: Vec<PgHeapTuple<'a, AllocatedByRust>>,
+    pub(crate) entries: pgrx::Array<'a, PgHeapTuple<'a, AllocatedByRust>>,
 }
 
 impl IntoDatum for CrunchyMap<'_> {
@@ -143,8 +141,27 @@ impl FromDatum for CrunchyMap<'_> {
         if is_null {
             None
         } else {
-            let entries = Vec::<PgHeapTuple<AllocatedByRust>>::from_datum(datum, is_null).unwrap();
+            let entries =
+                pgrx::Array::<PgHeapTuple<AllocatedByRust>>::from_datum(datum, false).unwrap();
+
             Some(CrunchyMap { entries })
         }
     }
 }
+
+unsafe impl<'a> UnboxDatum for CrunchyMap<'a> {
+    type As<'src> = Self
+    where
+        Self: 'src;
+
+    unsafe fn unbox<'src>(datum: pgrx::datum::Datum<'src>) -> Self::As<'src>
+    where
+        Self: 'src,
+    {
+        let entries =
+            pgrx::Array::<PgHeapTuple<AllocatedByRust>>::from_datum(datum.sans_lifetime(), false)
+                .unwrap();
+
+        CrunchyMap { entries }
+    }
+}
diff --git a/src/type_compat/pg_arrow_type_conversions.rs b/src/type_compat/pg_arrow_type_conversions.rs
index f12ee99..e70157e 100644
--- a/src/type_compat/pg_arrow_type_conversions.rs
+++ b/src/type_compat/pg_arrow_type_conversions.rs
@@ -1,10 +1,8 @@
 use std::ffi::CStr;
 
 use pgrx::{
-    direct_function_call,
-    pg_sys::{self, TimeTzADT},
-    AnyNumeric, Date, Interval, IntoDatum, Time, TimeWithTimeZone, Timestamp,
-    TimestampWithTimeZone,
+    datum::{Date, Interval, Time, TimeWithTimeZone, Timestamp, TimestampWithTimeZone},
+    direct_function_call, pg_sys, AnyNumeric, IntoDatum,
 };
 
 pub(crate) const MAX_DECIMAL_PRECISION: usize = 38;
@@ -48,7 +46,7 @@ pub(crate) fn timestamp_to_i64(timestamp: Timestamp) -> Option<i64> {
 }
 
 pub(crate) fn i64_to_timestamp(i64_timestamp: i64) -> Option<Timestamp> {
-    let timestamp: Timestamp = i64_timestamp.into();
+    let timestamp: Timestamp = i64_timestamp.try_into().unwrap();
 
     // Duckdb epoch is Unix epoch (1970-01-01). Convert it to PG epoch (2000-01-01). -10957 days
     let adjustment_interval = Interval::from_days(10957);
@@ -63,22 +61,6 @@ pub(crate) fn i64_to_timestamp(i64_timestamp: i64) -> Option<Timestamp> {
     Some(adjusted_timestamp)
 }
 
-pub(crate) fn i64_to_timestamptz(i64_timestamptz: i64) -> Option<TimestampWithTimeZone> {
-    let timestamptz: TimestampWithTimeZone = i64_timestamptz.try_into().unwrap();
-
-    // Duckdb epoch is Unix epoch (1970-01-01). Convert it to PG epoch (2000-01-01). -10957 days
-    let adjustment_interval = Interval::from_days(10957);
-    let adjusted_timestamptz: TimestampWithTimeZone = unsafe {
-        direct_function_call(
-            pg_sys::timestamptz_mi_interval,
-            &[timestamptz.into_datum(), adjustment_interval.into_datum()],
-        )
-        .unwrap()
-    };
-
-    Some(adjusted_timestamptz)
-}
-
 pub(crate) fn timestamptz_to_i64(timestamptz: TimestampWithTimeZone) -> Option<i64> {
     // PG epoch is (2000-01-01). Convert it to Unix epoch (1970-01-01). +10957 days
     let adjustment_interval = Interval::from_days(10957);
@@ -102,6 +84,22 @@ pub(crate) fn timestamptz_to_i64(timestamptz: TimestampWithTimeZone) -> Option<i
     ))
 }
 
+pub(crate) fn i64_to_timestamptz(i64_timestamptz: i64) -> Option<TimestampWithTimeZone> {
+    let timestamptz: TimestampWithTimeZone = i64_timestamptz.try_into().unwrap();
+
+    // Duckdb epoch is Unix epoch (1970-01-01). Convert it to PG epoch (2000-01-01). -10957 days
+    let adjustment_interval = Interval::from_days(10957);
+    let adjusted_timestamptz: TimestampWithTimeZone = unsafe {
+        direct_function_call(
+            pg_sys::timestamptz_mi_interval,
+            &[timestamptz.into_datum(), adjustment_interval.into_datum()],
+        )
+        .unwrap()
+    };
+
+    Some(adjusted_timestamptz)
+}
+
 pub(crate) fn time_to_i64(time: Time) -> Option<i64> {
     let time_as_bytes: Vec<u8> =
         unsafe { direct_function_call(pg_sys::time_send, &[time.into_datum()]).unwrap() };
@@ -109,8 +107,7 @@ pub(crate) fn time_to_i64(time: Time) -> Option<i64> {
 }
 
 pub(crate) fn i64_to_time(i64_time: i64) -> Option<Time> {
-    let time: Time = i64_time.into();
-    Some(time)
+    Some(i64_time.try_into().unwrap())
 }
 
 pub(crate) fn timetz_to_i64(timetz: TimeWithTimeZone) -> Option<i64> {
@@ -141,12 +138,7 @@ pub(crate) fn timetz_to_i64(timetz: TimeWithTimeZone) -> Option<i64> {
 }
 
 pub(crate) fn i64_to_timetz(i64_timetz: i64) -> Option<TimeWithTimeZone> {
-    let adjusted_timetz = TimeTzADT {
-        time: i64_timetz,
-        zone: 0,
-    };
-    let adjusted_timetz: TimeWithTimeZone = adjusted_timetz.into();
-    Some(adjusted_timetz)
+    i64_to_time(i64_timetz).map(|time| time.into())
 }
 
 pub(crate) fn numeric_to_i128(numeric: AnyNumeric) -> Option<i128> {