From 047ff2e51090556e2b358a3f521bcac453e5d5d8 Mon Sep 17 00:00:00 2001
From: Aykut Bozkurt <aykut.bozkurt1995@gmail.com>
Date: Mon, 23 Sep 2024 13:29:13 +0300
Subject: [PATCH] convert arrow array to Postgres rows

---
 src/arrow_parquet.rs                          |   1 +
 src/arrow_parquet/arrow_to_pg.rs              | 488 ++++++++++++++++++
 src/arrow_parquet/arrow_to_pg/bool.rs         |  30 ++
 src/arrow_parquet/arrow_to_pg/bytea.rs        |  33 ++
 src/arrow_parquet/arrow_to_pg/char.rs         |  34 ++
 src/arrow_parquet/arrow_to_pg/date.rs         |  33 ++
 .../arrow_to_pg/fallback_to_text.rs           |  38 ++
 src/arrow_parquet/arrow_to_pg/float4.rs       |  29 ++
 src/arrow_parquet/arrow_to_pg/float8.rs       |  29 ++
 src/arrow_parquet/arrow_to_pg/geometry.rs     |  38 ++
 src/arrow_parquet/arrow_to_pg/int2.rs         |  29 ++
 src/arrow_parquet/arrow_to_pg/int4.rs         |  29 ++
 src/arrow_parquet/arrow_to_pg/int8.rs         |  29 ++
 src/arrow_parquet/arrow_to_pg/map.rs          |  99 ++++
 src/arrow_parquet/arrow_to_pg/numeric.rs      |  37 ++
 src/arrow_parquet/arrow_to_pg/oid.rs          |  31 ++
 src/arrow_parquet/arrow_to_pg/record.rs       |  72 +++
 src/arrow_parquet/arrow_to_pg/text.rs         |  30 ++
 src/arrow_parquet/arrow_to_pg/time.rs         |  35 ++
 src/arrow_parquet/arrow_to_pg/timestamp.rs    |  37 ++
 src/arrow_parquet/arrow_to_pg/timestamptz.rs  |  39 ++
 src/arrow_parquet/arrow_to_pg/timetz.rs       |  37 ++
 22 files changed, 1257 insertions(+)
 create mode 100644 src/arrow_parquet/arrow_to_pg.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/bool.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/bytea.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/char.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/date.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/fallback_to_text.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/float4.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/float8.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/geometry.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/int2.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/int4.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/int8.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/map.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/numeric.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/oid.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/record.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/text.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/time.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/timestamp.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/timestamptz.rs
 create mode 100644 src/arrow_parquet/arrow_to_pg/timetz.rs

diff --git a/src/arrow_parquet.rs b/src/arrow_parquet.rs
index 460de91..e30f1fc 100644
--- a/src/arrow_parquet.rs
+++ b/src/arrow_parquet.rs
@@ -1,3 +1,4 @@
+pub(crate) mod arrow_to_pg;
 pub(crate) mod arrow_utils;
 pub(crate) mod codec;
 pub(crate) mod parquet_writer;
diff --git a/src/arrow_parquet/arrow_to_pg.rs b/src/arrow_parquet/arrow_to_pg.rs
new file mode 100644
index 0000000..608c248
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg.rs
@@ -0,0 +1,488 @@
+use arrow::array::{
+    Array, ArrayData, BinaryArray, BooleanArray, Date32Array, Decimal128Array, Float32Array,
+    Float64Array, Int16Array, Int32Array, Int64Array, ListArray, MapArray, StringArray,
+    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, IntoDatum, PgTupleDesc,
+};
+
+use crate::{
+    pgrx_utils::{array_element_typoid, is_array_type, is_composite_type, tuple_desc},
+    type_compat::{
+        fallback_to_text::{reset_fallback_to_text_context, FallbackToText},
+        geometry::{is_postgis_geometry_typoid, Geometry},
+        map::{is_crunchy_map_typoid, CrunchyMap},
+        pg_arrow_type_conversions::{
+            extract_precision_from_numeric_typmod, extract_scale_from_numeric_typmod,
+            MAX_DECIMAL_PRECISION,
+        },
+    },
+};
+
+pub(crate) mod bool;
+pub(crate) mod bytea;
+pub(crate) mod char;
+pub(crate) mod date;
+pub(crate) mod fallback_to_text;
+pub(crate) mod float4;
+pub(crate) mod float8;
+pub(crate) mod geometry;
+pub(crate) mod int2;
+pub(crate) mod int4;
+pub(crate) mod int8;
+pub(crate) mod map;
+pub(crate) mod numeric;
+pub(crate) mod oid;
+pub(crate) mod record;
+pub(crate) mod text;
+pub(crate) mod time;
+pub(crate) mod timestamp;
+pub(crate) mod timestamptz;
+pub(crate) mod timetz;
+
+pub(crate) trait ArrowArrayToPgType<'a, A: From<ArrayData>, T: 'a + IntoDatum> {
+    fn to_pg_type(array: A, context: ArrowToPgPerAttributeContext<'a>) -> Option<T>;
+}
+
+#[derive(Clone)]
+pub(crate) struct ArrowToPgPerAttributeContext<'a> {
+    typoid: Oid,
+    typmod: i32,
+    tupledesc: Option<PgTupleDesc<'a>>,
+    precision: Option<usize>,
+    scale: Option<usize>,
+}
+
+impl<'a> ArrowToPgPerAttributeContext<'a> {
+    pub(crate) fn new(typoid: Oid, typmod: i32) -> Self {
+        Self {
+            typoid,
+            typmod,
+            tupledesc: None,
+            precision: None,
+            scale: None,
+        }
+    }
+
+    pub(crate) fn with_tupledesc(mut self, tupledesc: PgTupleDesc<'a>) -> Self {
+        self.tupledesc = Some(tupledesc);
+        self
+    }
+
+    pub(crate) fn with_precision(mut self, precision: usize) -> Self {
+        self.precision = Some(precision);
+        self
+    }
+
+    pub(crate) fn with_scale(mut self, scale: usize) -> Self {
+        self.scale = Some(scale);
+        self
+    }
+}
+
+pub(crate) fn to_pg_datum(
+    attribute_array: ArrayData,
+    attribute_typoid: Oid,
+    attribute_typmod: i32,
+) -> Option<Datum> {
+    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 {
+        let attribute_context =
+            ArrowToPgPerAttributeContext::new(attribute_typoid, attribute_typmod);
+
+        to_pg_nonarray_datum(attribute_array, attribute_context)
+    }
+}
+
+fn to_pg_nonarray_datum(
+    primitive_array: ArrayData,
+    attribute_context: ArrowToPgPerAttributeContext,
+) -> Option<Datum> {
+    match attribute_context.typoid {
+        FLOAT4OID => {
+            let val = <f32 as ArrowArrayToPgType<Float32Array, f32>>::to_pg_type(
+                primitive_array.into(),
+                attribute_context,
+            );
+            val.into_datum()
+        }
+        FLOAT8OID => {
+            let val = <f64 as ArrowArrayToPgType<Float64Array, f64>>::to_pg_type(
+                primitive_array.into(),
+                attribute_context,
+            );
+            val.into_datum()
+        }
+        INT2OID => {
+            let val = <i16 as ArrowArrayToPgType<Int16Array, i16>>::to_pg_type(
+                primitive_array.into(),
+                attribute_context,
+            );
+            val.into_datum()
+        }
+        INT4OID => {
+            let val = <i32 as ArrowArrayToPgType<Int32Array, i32>>::to_pg_type(
+                primitive_array.into(),
+                attribute_context,
+            );
+            val.into_datum()
+        }
+        INT8OID => {
+            let val = <i64 as ArrowArrayToPgType<Int64Array, i64>>::to_pg_type(
+                primitive_array.into(),
+                attribute_context,
+            );
+            val.into_datum()
+        }
+        BOOLOID => {
+            let val = <bool as ArrowArrayToPgType<BooleanArray, bool>>::to_pg_type(
+                primitive_array.into(),
+                attribute_context,
+            );
+            val.into_datum()
+        }
+        CHAROID => {
+            let val = <i8 as ArrowArrayToPgType<StringArray, i8>>::to_pg_type(
+                primitive_array.into(),
+                attribute_context,
+            );
+            val.into_datum()
+        }
+        TEXTOID => {
+            let val = <String as ArrowArrayToPgType<StringArray, String>>::to_pg_type(
+                primitive_array.into(),
+                attribute_context,
+            );
+            val.into_datum()
+        }
+        BYTEAOID => {
+            let val = <Vec<u8> as ArrowArrayToPgType<BinaryArray, Vec<u8>>>::to_pg_type(
+                primitive_array.into(),
+                attribute_context,
+            );
+            val.into_datum()
+        }
+        OIDOID => {
+            let val = <Oid as ArrowArrayToPgType<UInt32Array, Oid>>::to_pg_type(
+                primitive_array.into(),
+                attribute_context,
+            );
+            val.into_datum()
+        }
+        NUMERICOID => to_pg_numeric_datum(primitive_array, attribute_context),
+        DATEOID => {
+            let val = <Date as ArrowArrayToPgType<Date32Array, Date>>::to_pg_type(
+                primitive_array.into(),
+                attribute_context,
+            );
+            val.into_datum()
+        }
+        TIMEOID => {
+            let val = <Time as ArrowArrayToPgType<Time64MicrosecondArray, Time>>::to_pg_type(
+                primitive_array.into(),
+                attribute_context,
+            );
+            val.into_datum()
+        }
+        TIMETZOID => {
+            let val = <TimeWithTimeZone as ArrowArrayToPgType<
+                Time64MicrosecondArray,
+                TimeWithTimeZone,
+            >>::to_pg_type(primitive_array.into(), attribute_context);
+            val.into_datum()
+        }
+        TIMESTAMPOID => {
+            let val =
+                <Timestamp as ArrowArrayToPgType<TimestampMicrosecondArray, Timestamp>>::to_pg_type(
+                    primitive_array.into(),
+                    attribute_context,
+                );
+            val.into_datum()
+        }
+        TIMESTAMPTZOID => {
+            let val = <TimestampWithTimeZone as ArrowArrayToPgType<
+                TimestampMicrosecondArray,
+                TimestampWithTimeZone,
+            >>::to_pg_type(primitive_array.into(), attribute_context);
+            val.into_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);
+
+                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 {
+                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()
+            }
+        }
+    }
+}
+
+fn to_pg_array_datum(
+    list_array: ListArray,
+    attribute_context: ArrowToPgPerAttributeContext,
+) -> Option<Datum> {
+    if list_array.is_null(0) {
+        return None;
+    }
+
+    let list_array = list_array.value(0).to_data();
+
+    match attribute_context.typoid {
+        FLOAT4OID => {
+            let val =
+                <Vec<Option<f32>> as ArrowArrayToPgType<Float32Array, Vec<Option<f32>>>>::to_pg_type(
+                    list_array.into(),
+                    attribute_context
+                );
+            val.into_datum()
+        }
+        FLOAT8OID => {
+            let val =
+                <Vec<Option<f64>> as ArrowArrayToPgType<Float64Array, Vec<Option<f64>>>>::to_pg_type(
+                    list_array.into(),
+                    attribute_context
+                );
+            val.into_datum()
+        }
+        INT2OID => {
+            let val =
+                <Vec<Option<i16>> as ArrowArrayToPgType<Int16Array, Vec<Option<i16>>>>::to_pg_type(
+                    list_array.into(),
+                    attribute_context,
+                );
+            val.into_datum()
+        }
+        INT4OID => {
+            let val =
+                <Vec<Option<i32>> as ArrowArrayToPgType<Int32Array, Vec<Option<i32>>>>::to_pg_type(
+                    list_array.into(),
+                    attribute_context,
+                );
+            val.into_datum()
+        }
+        INT8OID => {
+            let val =
+                <Vec<Option<i64>> as ArrowArrayToPgType<Int64Array, Vec<Option<i64>>>>::to_pg_type(
+                    list_array.into(),
+                    attribute_context,
+                );
+            val.into_datum()
+        }
+        BOOLOID => {
+            let val =
+                <Vec<Option<bool>> as ArrowArrayToPgType<BooleanArray, Vec<Option<bool>>>>::to_pg_type(
+                    list_array.into(),
+                    attribute_context
+                );
+            val.into_datum()
+        }
+        CHAROID => {
+            let val =
+                <Vec<Option<i8>> as ArrowArrayToPgType<StringArray, Vec<Option<i8>>>>::to_pg_type(
+                    list_array.into(),
+                    attribute_context,
+                );
+            val.into_datum()
+        }
+        TEXTOID => {
+            let val = <Vec<Option<String>> as ArrowArrayToPgType<
+                StringArray,
+                Vec<Option<String>>,
+            >>::to_pg_type(list_array.into(), attribute_context);
+            val.into_datum()
+        }
+        BYTEAOID => {
+            let val = <Vec<Option<Vec<u8>>> as ArrowArrayToPgType<
+                BinaryArray,
+                Vec<Option<Vec<u8>>>,
+            >>::to_pg_type(list_array.into(), attribute_context);
+            val.into_datum()
+        }
+        OIDOID => {
+            let val =
+                <Vec<Option<Oid>> as ArrowArrayToPgType<UInt32Array, Vec<Option<Oid>>>>::to_pg_type(
+                    list_array.into(),
+                    attribute_context,
+                );
+            val.into_datum()
+        }
+        NUMERICOID => to_pg_numeric_array_datum(list_array, attribute_context),
+        DATEOID => {
+            let val =
+                <Vec<Option<Date>> as ArrowArrayToPgType<Date32Array, Vec<Option<Date>>>>::to_pg_type(
+                    list_array.into(),
+                    attribute_context
+                );
+            val.into_datum()
+        }
+        TIMEOID => {
+            let val = <Vec<Option<Time>> as ArrowArrayToPgType<
+                Time64MicrosecondArray,
+                Vec<Option<Time>>,
+            >>::to_pg_type(list_array.into(), attribute_context);
+            val.into_datum()
+        }
+        TIMETZOID => {
+            let val = <Vec<Option<TimeWithTimeZone>> as ArrowArrayToPgType<
+                Time64MicrosecondArray,
+                Vec<Option<TimeWithTimeZone>>,
+            >>::to_pg_type(list_array.into(), attribute_context);
+            val.into_datum()
+        }
+        TIMESTAMPOID => {
+            let val = <Vec<Option<Timestamp>> as ArrowArrayToPgType<
+                TimestampMicrosecondArray,
+                Vec<Option<Timestamp>>,
+            >>::to_pg_type(list_array.into(), attribute_context);
+            val.into_datum()
+        }
+        TIMESTAMPTZOID => {
+            let val = <Vec<Option<TimestampWithTimeZone>> as ArrowArrayToPgType<
+                TimestampMicrosecondArray,
+                Vec<Option<TimestampWithTimeZone>>,
+            >>::to_pg_type(list_array.into(), attribute_context);
+            val.into_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);
+
+                let val = <Vec<Option<PgHeapTuple<AllocatedByRust>>> as ArrowArrayToPgType<
+                    StructArray,
+                    Vec<Option<PgHeapTuple<AllocatedByRust>>>,
+                >>::to_pg_type(list_array.into(), attribute_context);
+
+                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);
+
+                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(list_array.into(), attribute_context);
+
+                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(list_array.into(), attribute_context);
+
+                val.into_datum()
+            }
+        }
+    }
+}
+
+fn to_pg_numeric_datum(
+    numeric_array: ArrayData,
+    attribute_context: ArrowToPgPerAttributeContext,
+) -> Option<Datum> {
+    let precision = extract_precision_from_numeric_typmod(attribute_context.typmod);
+
+    if precision > MAX_DECIMAL_PRECISION {
+        reset_fallback_to_text_context(attribute_context.typoid, attribute_context.typmod);
+
+        let val = <FallbackToText as ArrowArrayToPgType<StringArray, FallbackToText>>::to_pg_type(
+            numeric_array.into(),
+            attribute_context,
+        );
+
+        val.into_datum()
+    } else {
+        let scale = extract_scale_from_numeric_typmod(attribute_context.typmod);
+
+        let val = <AnyNumeric as ArrowArrayToPgType<Decimal128Array, AnyNumeric>>::to_pg_type(
+            numeric_array.into(),
+            attribute_context
+                .with_precision(precision)
+                .with_scale(scale),
+        );
+
+        val.into_datum()
+    }
+}
+
+fn to_pg_numeric_array_datum(
+    numeric_list_array: ArrayData,
+    attribute_context: ArrowToPgPerAttributeContext,
+) -> Option<Datum> {
+    let precision = extract_precision_from_numeric_typmod(attribute_context.typmod);
+
+    if precision > MAX_DECIMAL_PRECISION {
+        reset_fallback_to_text_context(attribute_context.typoid, attribute_context.typmod);
+
+        let val = <Vec<Option<FallbackToText>> as ArrowArrayToPgType<
+            StringArray,
+            Vec<Option<FallbackToText>>,
+        >>::to_pg_type(numeric_list_array.into(), attribute_context);
+
+        val.into_datum()
+    } else {
+        let scale = extract_scale_from_numeric_typmod(attribute_context.typmod);
+
+        let val = <Vec<Option<AnyNumeric>> as ArrowArrayToPgType<
+            Decimal128Array,
+            Vec<Option<AnyNumeric>>,
+        >>::to_pg_type(
+            numeric_list_array.into(),
+            attribute_context
+                .with_precision(precision)
+                .with_scale(scale),
+        );
+
+        val.into_datum()
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/bool.rs b/src/arrow_parquet/arrow_to_pg/bool.rs
new file mode 100644
index 0000000..dd6a447
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/bool.rs
@@ -0,0 +1,30 @@
+use arrow::array::{Array, BooleanArray};
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Bool
+impl ArrowArrayToPgType<'_, BooleanArray, bool> for bool {
+    fn to_pg_type(arr: BooleanArray, _context: ArrowToPgPerAttributeContext<'_>) -> Option<bool> {
+        if arr.is_null(0) {
+            None
+        } else {
+            let val = arr.value(0);
+            Some(val)
+        }
+    }
+}
+
+// Bool[]
+impl ArrowArrayToPgType<'_, BooleanArray, Vec<Option<bool>>> for Vec<Option<bool>> {
+    fn to_pg_type(
+        arr: BooleanArray,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<bool>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            vals.push(val);
+        }
+
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/bytea.rs b/src/arrow_parquet/arrow_to_pg/bytea.rs
new file mode 100644
index 0000000..001aa8b
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/bytea.rs
@@ -0,0 +1,33 @@
+use arrow::array::{Array, BinaryArray};
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Bytea
+impl ArrowArrayToPgType<'_, BinaryArray, Vec<u8>> for Vec<u8> {
+    fn to_pg_type(arr: BinaryArray, _context: ArrowToPgPerAttributeContext<'_>) -> Option<Vec<u8>> {
+        if arr.is_null(0) {
+            None
+        } else {
+            Some(arr.value(0).to_vec())
+        }
+    }
+}
+
+// Bytea[]
+impl ArrowArrayToPgType<'_, BinaryArray, Vec<Option<Vec<u8>>>> for Vec<Option<Vec<u8>>> {
+    fn to_pg_type(
+        arr: BinaryArray,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<Vec<u8>>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            if let Some(val) = val {
+                vals.push(Some(val.to_vec()));
+            } else {
+                vals.push(None);
+            }
+        }
+
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/char.rs b/src/arrow_parquet/arrow_to_pg/char.rs
new file mode 100644
index 0000000..c7c6dbe
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/char.rs
@@ -0,0 +1,34 @@
+use arrow::array::{Array, StringArray};
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Char
+impl ArrowArrayToPgType<'_, StringArray, i8> for i8 {
+    fn to_pg_type(arr: StringArray, _context: ArrowToPgPerAttributeContext<'_>) -> Option<i8> {
+        if arr.is_null(0) {
+            None
+        } else {
+            let val = arr.value(0);
+            let val: i8 = val.chars().next().expect("unexpected ascii char") as i8;
+            Some(val)
+        }
+    }
+}
+
+// Char[]
+impl ArrowArrayToPgType<'_, StringArray, Vec<Option<i8>>> for Vec<Option<i8>> {
+    fn to_pg_type(
+        arr: StringArray,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<i8>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            let val = val.map(|val| {
+                let val: i8 = val.chars().next().expect("unexpected ascii char") as i8;
+                val
+            });
+            vals.push(val);
+        }
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/date.rs b/src/arrow_parquet/arrow_to_pg/date.rs
new file mode 100644
index 0000000..001d893
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/date.rs
@@ -0,0 +1,33 @@
+use arrow::array::{Array, Date32Array};
+use pgrx::datum::Date;
+
+use crate::type_compat::pg_arrow_type_conversions::i32_to_date;
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Date
+impl ArrowArrayToPgType<'_, Date32Array, Date> for Date {
+    fn to_pg_type(arr: Date32Array, _context: ArrowToPgPerAttributeContext<'_>) -> Option<Date> {
+        if arr.is_null(0) {
+            None
+        } else {
+            let val = arr.value(0);
+            Some(i32_to_date(val))
+        }
+    }
+}
+
+// Date[]
+impl ArrowArrayToPgType<'_, Date32Array, Vec<Option<Date>>> for Vec<Option<Date>> {
+    fn to_pg_type(
+        arr: Date32Array,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<Date>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            let val = val.map(i32_to_date);
+            vals.push(val);
+        }
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/fallback_to_text.rs b/src/arrow_parquet/arrow_to_pg/fallback_to_text.rs
new file mode 100644
index 0000000..0393fd9
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/fallback_to_text.rs
@@ -0,0 +1,38 @@
+use arrow::array::{Array, StringArray};
+
+use crate::type_compat::fallback_to_text::FallbackToText;
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Text representation of any type
+impl ArrowArrayToPgType<'_, StringArray, FallbackToText> for FallbackToText {
+    fn to_pg_type(
+        arr: StringArray,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<FallbackToText> {
+        if arr.is_null(0) {
+            None
+        } else {
+            let text_repr = arr.value(0).to_string();
+            let val = FallbackToText(text_repr);
+            Some(val)
+        }
+    }
+}
+
+// Text[] representation of any type
+impl ArrowArrayToPgType<'_, StringArray, Vec<Option<FallbackToText>>>
+    for Vec<Option<FallbackToText>>
+{
+    fn to_pg_type(
+        arr: StringArray,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<FallbackToText>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            let val = val.map(|val| FallbackToText(val.to_string()));
+            vals.push(val);
+        }
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/float4.rs b/src/arrow_parquet/arrow_to_pg/float4.rs
new file mode 100644
index 0000000..0142c53
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/float4.rs
@@ -0,0 +1,29 @@
+use arrow::array::{Array, Float32Array};
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Float4
+impl ArrowArrayToPgType<'_, Float32Array, f32> for f32 {
+    fn to_pg_type(arr: Float32Array, _context: ArrowToPgPerAttributeContext<'_>) -> Option<f32> {
+        if arr.is_null(0) {
+            None
+        } else {
+            let val = arr.value(0);
+            Some(val)
+        }
+    }
+}
+
+// Float4[]
+impl ArrowArrayToPgType<'_, Float32Array, Vec<Option<f32>>> for Vec<Option<f32>> {
+    fn to_pg_type(
+        arr: Float32Array,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<f32>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            vals.push(val);
+        }
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/float8.rs b/src/arrow_parquet/arrow_to_pg/float8.rs
new file mode 100644
index 0000000..a847377
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/float8.rs
@@ -0,0 +1,29 @@
+use arrow::array::{Array, Float64Array};
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Float8
+impl ArrowArrayToPgType<'_, Float64Array, f64> for f64 {
+    fn to_pg_type(arr: Float64Array, _context: ArrowToPgPerAttributeContext<'_>) -> Option<f64> {
+        if arr.is_null(0) {
+            None
+        } else {
+            let val = arr.value(0);
+            Some(val)
+        }
+    }
+}
+
+// Float8[]
+impl ArrowArrayToPgType<'_, Float64Array, Vec<Option<f64>>> for Vec<Option<f64>> {
+    fn to_pg_type(
+        arr: Float64Array,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<f64>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            vals.push(val);
+        }
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/geometry.rs b/src/arrow_parquet/arrow_to_pg/geometry.rs
new file mode 100644
index 0000000..b06d5bb
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/geometry.rs
@@ -0,0 +1,38 @@
+use arrow::array::{Array, BinaryArray};
+
+use crate::type_compat::geometry::Geometry;
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Geometry
+impl ArrowArrayToPgType<'_, BinaryArray, Geometry> for Geometry {
+    fn to_pg_type(
+        arr: BinaryArray,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Geometry> {
+        if arr.is_null(0) {
+            None
+        } else {
+            Some(arr.value(0).to_vec().into())
+        }
+    }
+}
+
+// Geometry[]
+impl ArrowArrayToPgType<'_, BinaryArray, Vec<Option<Geometry>>> for Vec<Option<Geometry>> {
+    fn to_pg_type(
+        arr: BinaryArray,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<Geometry>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            if let Some(val) = val {
+                vals.push(Some(val.to_vec().into()));
+            } else {
+                vals.push(None);
+            }
+        }
+
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/int2.rs b/src/arrow_parquet/arrow_to_pg/int2.rs
new file mode 100644
index 0000000..89e15d2
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/int2.rs
@@ -0,0 +1,29 @@
+use arrow::array::{Array, Int16Array};
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Int2
+impl ArrowArrayToPgType<'_, Int16Array, i16> for i16 {
+    fn to_pg_type(arr: Int16Array, _context: ArrowToPgPerAttributeContext<'_>) -> Option<i16> {
+        if arr.is_null(0) {
+            None
+        } else {
+            let val = arr.value(0);
+            Some(val)
+        }
+    }
+}
+
+// Int2[]
+impl ArrowArrayToPgType<'_, Int16Array, Vec<Option<i16>>> for Vec<Option<i16>> {
+    fn to_pg_type(
+        arr: Int16Array,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<i16>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            vals.push(val);
+        }
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/int4.rs b/src/arrow_parquet/arrow_to_pg/int4.rs
new file mode 100644
index 0000000..19de839
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/int4.rs
@@ -0,0 +1,29 @@
+use arrow::array::{Array, Int32Array};
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Int4
+impl ArrowArrayToPgType<'_, Int32Array, i32> for i32 {
+    fn to_pg_type(arr: Int32Array, _context: ArrowToPgPerAttributeContext<'_>) -> Option<i32> {
+        if arr.is_null(0) {
+            None
+        } else {
+            let val = arr.value(0);
+            Some(val)
+        }
+    }
+}
+
+// Int4[]
+impl ArrowArrayToPgType<'_, Int32Array, Vec<Option<i32>>> for Vec<Option<i32>> {
+    fn to_pg_type(
+        arr: Int32Array,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<i32>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            vals.push(val);
+        }
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/int8.rs b/src/arrow_parquet/arrow_to_pg/int8.rs
new file mode 100644
index 0000000..306597f
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/int8.rs
@@ -0,0 +1,29 @@
+use arrow::array::{Array, Int64Array};
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Int8
+impl ArrowArrayToPgType<'_, Int64Array, i64> for i64 {
+    fn to_pg_type(arr: Int64Array, _context: ArrowToPgPerAttributeContext<'_>) -> Option<i64> {
+        if arr.is_null(0) {
+            None
+        } else {
+            let val = arr.value(0);
+            Some(val)
+        }
+    }
+}
+
+// Int8[]
+impl ArrowArrayToPgType<'_, Int64Array, Vec<Option<i64>>> for Vec<Option<i64>> {
+    fn to_pg_type(
+        arr: Int64Array,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<i64>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            vals.push(val);
+        }
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/map.rs b/src/arrow_parquet/arrow_to_pg/map.rs
new file mode 100644
index 0000000..0912916
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/map.rs
@@ -0,0 +1,99 @@
+use arrow::array::{Array, MapArray, StructArray};
+use pgrx::{prelude::PgHeapTuple, AllocatedByRust, FromDatum, IntoDatum};
+
+use crate::{
+    pgrx_utils::{domain_array_base_elem_typoid, tuple_desc},
+    type_compat::map::CrunchyMap,
+};
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// crunchy_map.key_<type1>_val_<type2>
+impl<'b, 'a: 'b> ArrowArrayToPgType<'b, MapArray, CrunchyMap<'a>> for CrunchyMap<'a> {
+    fn to_pg_type(
+        arr: MapArray,
+        context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<CrunchyMap<'a>> {
+        if arr.is_null(0) {
+            None
+        } else {
+            let entries_array = arr.value(0);
+
+            let entries_typoid = domain_array_base_elem_typoid(context.typoid);
+
+            let entries_tupledesc = tuple_desc(entries_typoid, context.typmod);
+
+            let entries_context = ArrowToPgPerAttributeContext::new(entries_typoid, context.typmod)
+                .with_tupledesc(entries_tupledesc);
+
+            let entries = <Vec<Option<PgHeapTuple<AllocatedByRust>>> as ArrowArrayToPgType<
+                StructArray,
+                Vec<Option<PgHeapTuple<AllocatedByRust>>>,
+            >>::to_pg_type(entries_array, entries_context);
+
+            if let Some(entries) = entries {
+                let entries_datum = entries.into_datum();
+
+                if let Some(entries_datum) = entries_datum {
+                    let entries = unsafe {
+                        pgrx::Array::from_datum(entries_datum, false)
+                            .expect("map entries should be an array")
+                    };
+                    Some(CrunchyMap { entries })
+                } else {
+                    None
+                }
+            } else {
+                None
+            }
+        }
+    }
+}
+
+// crunchy_map.key_<type1>_val_<type2>[]
+impl<'b, 'a: 'b> ArrowArrayToPgType<'b, MapArray, Vec<Option<CrunchyMap<'a>>>>
+    for Vec<Option<CrunchyMap<'a>>>
+{
+    fn to_pg_type(
+        array: MapArray,
+        context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<CrunchyMap<'a>>>> {
+        let mut maps = vec![];
+
+        let entries_typoid = domain_array_base_elem_typoid(context.typoid);
+
+        let entries_tupledesc = tuple_desc(entries_typoid, context.typmod);
+
+        let entries_context = ArrowToPgPerAttributeContext::new(entries_typoid, context.typmod)
+            .with_tupledesc(entries_tupledesc);
+
+        for entries_array in array.iter() {
+            if let Some(entries_array) = entries_array {
+                let entries = <Vec<Option<PgHeapTuple<AllocatedByRust>>> as ArrowArrayToPgType<
+                    StructArray,
+                    Vec<Option<PgHeapTuple<AllocatedByRust>>>,
+                >>::to_pg_type(entries_array, entries_context.clone());
+
+                if let Some(entries) = entries {
+                    let entries_datum = entries.into_datum();
+
+                    if let Some(entries_datum) = entries_datum {
+                        let entries = unsafe {
+                            pgrx::Array::from_datum(entries_datum, false)
+                                .expect("map entries should be an array")
+                        };
+                        maps.push(Some(CrunchyMap { entries }))
+                    } else {
+                        maps.push(None);
+                    }
+                } else {
+                    maps.push(None);
+                }
+            } else {
+                maps.push(None);
+            }
+        }
+
+        Some(maps)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/numeric.rs b/src/arrow_parquet/arrow_to_pg/numeric.rs
new file mode 100644
index 0000000..3739256
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/numeric.rs
@@ -0,0 +1,37 @@
+use arrow::array::{Array, Decimal128Array};
+use pgrx::AnyNumeric;
+
+use crate::type_compat::pg_arrow_type_conversions::i128_to_numeric;
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Numeric
+impl ArrowArrayToPgType<'_, Decimal128Array, AnyNumeric> for AnyNumeric {
+    fn to_pg_type(
+        arr: Decimal128Array,
+        context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<AnyNumeric> {
+        if arr.is_null(0) {
+            None
+        } else {
+            let scale = context.scale.expect("Expected scale");
+            Some(i128_to_numeric(arr.value(0), scale))
+        }
+    }
+}
+
+// Numeric[]
+impl ArrowArrayToPgType<'_, Decimal128Array, Vec<Option<AnyNumeric>>> for Vec<Option<AnyNumeric>> {
+    fn to_pg_type(
+        arr: Decimal128Array,
+        context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<AnyNumeric>>> {
+        let scale = context.scale.expect("Expected scale");
+        let mut vals = vec![];
+        for val in arr.iter() {
+            let val = val.map(|v| i128_to_numeric(v, scale));
+            vals.push(val);
+        }
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/oid.rs b/src/arrow_parquet/arrow_to_pg/oid.rs
new file mode 100644
index 0000000..0cd0ba3
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/oid.rs
@@ -0,0 +1,31 @@
+use arrow::array::{Array, UInt32Array};
+use pgrx::pg_sys::Oid;
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Oid
+impl ArrowArrayToPgType<'_, UInt32Array, Oid> for Oid {
+    fn to_pg_type(arr: UInt32Array, _context: ArrowToPgPerAttributeContext<'_>) -> Option<Oid> {
+        if arr.is_null(0) {
+            None
+        } else {
+            let val = arr.value(0);
+            Some(val.into())
+        }
+    }
+}
+
+// Oid[]
+impl ArrowArrayToPgType<'_, UInt32Array, Vec<Option<Oid>>> for Vec<Option<Oid>> {
+    fn to_pg_type(
+        arr: UInt32Array,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<Oid>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            let val = val.map(|val| val.into());
+            vals.push(val);
+        }
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/record.rs b/src/arrow_parquet/arrow_to_pg/record.rs
new file mode 100644
index 0000000..e21dc7e
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/record.rs
@@ -0,0 +1,72 @@
+use arrow::array::{Array, StructArray};
+use pgrx::{prelude::PgHeapTuple, AllocatedByRust};
+
+use crate::pgrx_utils::collect_valid_attributes;
+
+use super::{to_pg_datum, ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// PgHeapTuple
+impl<'a> ArrowArrayToPgType<'a, StructArray, PgHeapTuple<'a, AllocatedByRust>>
+    for PgHeapTuple<'a, AllocatedByRust>
+{
+    fn to_pg_type(
+        arr: StructArray,
+        context: ArrowToPgPerAttributeContext<'a>,
+    ) -> Option<PgHeapTuple<'a, AllocatedByRust>> {
+        if arr.is_null(0) {
+            return None;
+        }
+
+        let tupledesc = context.tupledesc.expect("Expected tupledesc");
+
+        let mut datums = vec![];
+
+        let include_generated_columns = false;
+        let attributes = collect_valid_attributes(&tupledesc, include_generated_columns);
+
+        for attribute in attributes {
+            let name = attribute.name();
+            let typoid = attribute.type_oid().value();
+            let typmod = attribute.type_mod();
+
+            let column_data = arr
+                .column_by_name(name)
+                .unwrap_or_else(|| panic!("column {} not found", name));
+
+            let datum = to_pg_datum(column_data.into_data(), typoid, typmod);
+            datums.push(datum);
+        }
+
+        Some(
+            unsafe { PgHeapTuple::from_datums(tupledesc, datums) }.unwrap_or_else(|e| {
+                panic!("failed to create heap tuple: {}", e);
+            }),
+        )
+    }
+}
+
+// PgHeapTuple[]
+impl<'a> ArrowArrayToPgType<'a, StructArray, Vec<Option<PgHeapTuple<'a, AllocatedByRust>>>>
+    for Vec<Option<PgHeapTuple<'a, AllocatedByRust>>>
+{
+    fn to_pg_type(
+        arr: StructArray,
+        context: ArrowToPgPerAttributeContext<'a>,
+    ) -> Option<Vec<Option<PgHeapTuple<'a, AllocatedByRust>>>> {
+        let len = arr.len();
+        let mut values = Vec::with_capacity(len);
+
+        for i in 0..len {
+            let tuple = arr.slice(i, 1);
+
+            let tuple = <PgHeapTuple<AllocatedByRust> as ArrowArrayToPgType<
+                StructArray,
+                PgHeapTuple<AllocatedByRust>,
+            >>::to_pg_type(tuple, context.clone());
+
+            values.push(tuple);
+        }
+
+        Some(values)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/text.rs b/src/arrow_parquet/arrow_to_pg/text.rs
new file mode 100644
index 0000000..3c70022
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/text.rs
@@ -0,0 +1,30 @@
+use arrow::array::{Array, StringArray};
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Text
+impl ArrowArrayToPgType<'_, StringArray, String> for String {
+    fn to_pg_type(arr: StringArray, _context: ArrowToPgPerAttributeContext<'_>) -> Option<String> {
+        if arr.is_null(0) {
+            None
+        } else {
+            let val = arr.value(0);
+            Some(val.to_string())
+        }
+    }
+}
+
+// Text[]
+impl ArrowArrayToPgType<'_, StringArray, Vec<Option<String>>> for Vec<Option<String>> {
+    fn to_pg_type(
+        arr: StringArray,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<String>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            let val = val.map(|val| val.to_string());
+            vals.push(val);
+        }
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/time.rs b/src/arrow_parquet/arrow_to_pg/time.rs
new file mode 100644
index 0000000..1f36f88
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/time.rs
@@ -0,0 +1,35 @@
+use arrow::array::{Array, Time64MicrosecondArray};
+use pgrx::datum::Time;
+
+use crate::type_compat::pg_arrow_type_conversions::i64_to_time;
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Time
+impl ArrowArrayToPgType<'_, Time64MicrosecondArray, Time> for Time {
+    fn to_pg_type(
+        arr: Time64MicrosecondArray,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Time> {
+        if arr.is_null(0) {
+            None
+        } else {
+            Some(i64_to_time(arr.value(0)))
+        }
+    }
+}
+
+// Time[]
+impl ArrowArrayToPgType<'_, Time64MicrosecondArray, Vec<Option<Time>>> for Vec<Option<Time>> {
+    fn to_pg_type(
+        arr: Time64MicrosecondArray,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<Time>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            let val = val.map(i64_to_time);
+            vals.push(val);
+        }
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/timestamp.rs b/src/arrow_parquet/arrow_to_pg/timestamp.rs
new file mode 100644
index 0000000..dd05afa
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/timestamp.rs
@@ -0,0 +1,37 @@
+use arrow::array::{Array, TimestampMicrosecondArray};
+use pgrx::datum::Timestamp;
+
+use crate::type_compat::pg_arrow_type_conversions::i64_to_timestamp;
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Timestamp
+impl ArrowArrayToPgType<'_, TimestampMicrosecondArray, Timestamp> for Timestamp {
+    fn to_pg_type(
+        arr: TimestampMicrosecondArray,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Timestamp> {
+        if arr.is_null(0) {
+            None
+        } else {
+            Some(i64_to_timestamp(arr.value(0)))
+        }
+    }
+}
+
+// Timestamp[]
+impl ArrowArrayToPgType<'_, TimestampMicrosecondArray, Vec<Option<Timestamp>>>
+    for Vec<Option<Timestamp>>
+{
+    fn to_pg_type(
+        arr: TimestampMicrosecondArray,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<Timestamp>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            let val = val.map(i64_to_timestamp);
+            vals.push(val);
+        }
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/timestamptz.rs b/src/arrow_parquet/arrow_to_pg/timestamptz.rs
new file mode 100644
index 0000000..4f76763
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/timestamptz.rs
@@ -0,0 +1,39 @@
+use arrow::array::{Array, TimestampMicrosecondArray};
+use pgrx::datum::TimestampWithTimeZone;
+
+use crate::type_compat::pg_arrow_type_conversions::i64_to_timestamptz;
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Timestamptz
+impl ArrowArrayToPgType<'_, TimestampMicrosecondArray, TimestampWithTimeZone>
+    for TimestampWithTimeZone
+{
+    fn to_pg_type(
+        arr: TimestampMicrosecondArray,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<TimestampWithTimeZone> {
+        if arr.is_null(0) {
+            None
+        } else {
+            Some(i64_to_timestamptz(arr.value(0)))
+        }
+    }
+}
+
+// Timestamptz[]
+impl ArrowArrayToPgType<'_, TimestampMicrosecondArray, Vec<Option<TimestampWithTimeZone>>>
+    for Vec<Option<TimestampWithTimeZone>>
+{
+    fn to_pg_type(
+        arr: TimestampMicrosecondArray,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<TimestampWithTimeZone>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            let val = val.map(i64_to_timestamptz);
+            vals.push(val);
+        }
+        Some(vals)
+    }
+}
diff --git a/src/arrow_parquet/arrow_to_pg/timetz.rs b/src/arrow_parquet/arrow_to_pg/timetz.rs
new file mode 100644
index 0000000..3397bc0
--- /dev/null
+++ b/src/arrow_parquet/arrow_to_pg/timetz.rs
@@ -0,0 +1,37 @@
+use arrow::array::{Array, Time64MicrosecondArray};
+use pgrx::datum::TimeWithTimeZone;
+
+use crate::type_compat::pg_arrow_type_conversions::i64_to_timetz;
+
+use super::{ArrowArrayToPgType, ArrowToPgPerAttributeContext};
+
+// Timetz
+impl ArrowArrayToPgType<'_, Time64MicrosecondArray, TimeWithTimeZone> for TimeWithTimeZone {
+    fn to_pg_type(
+        arr: Time64MicrosecondArray,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<TimeWithTimeZone> {
+        if arr.is_null(0) {
+            None
+        } else {
+            Some(i64_to_timetz(arr.value(0)))
+        }
+    }
+}
+
+// Timetz[]
+impl ArrowArrayToPgType<'_, Time64MicrosecondArray, Vec<Option<TimeWithTimeZone>>>
+    for Vec<Option<TimeWithTimeZone>>
+{
+    fn to_pg_type(
+        arr: Time64MicrosecondArray,
+        _context: ArrowToPgPerAttributeContext<'_>,
+    ) -> Option<Vec<Option<TimeWithTimeZone>>> {
+        let mut vals = vec![];
+        for val in arr.iter() {
+            let val = val.map(i64_to_timetz);
+            vals.push(val);
+        }
+        Some(vals)
+    }
+}