PostgreSQL interval type

Signed-off-by: itowlson <[email protected]>

Author: itowlson

Cargo.lock

@@ -6,6 +6,7 @@ edition = { workspace = true }
 
 [dependencies]
 anyhow = { workspace = true }
+bytes = {workspace = true }
 chrono = { workspace = true }
 deadpool-postgres = { version = "0.14", features = ["rt_tokio_1"] }
 moka = { version = "0.12", features = ["sync"] }

crates/factor-outbound-pg/Cargo.toml

@@ -5,8 +5,8 @@ use spin_world::async_trait;
 use spin_world::spin::postgres4_0_0::postgres::{
     self as v4, Column, DbDataType, DbValue, ParameterValue, RowSet,
 };
-use tokio_postgres::types::Type;
-use tokio_postgres::{config::SslMode, types::ToSql, NoTls, Row};
+use tokio_postgres::types::{FromSql, ToSql, Type};
+use tokio_postgres::{config::SslMode, NoTls, Row};
 
 /// Max connections in a given address' connection pool
 const CONNECTION_POOL_SIZE: usize = 64;
@@ -234,6 +234,7 @@ fn to_sql_parameter(value: &ParameterValue) -> Result<Box<dyn ToSql + Send + Syn
         ParameterValue::ArrayInt32(vs) => Ok(Box::new(vs.to_owned())),
         ParameterValue::ArrayInt64(vs) => Ok(Box::new(vs.to_owned())),
         ParameterValue::ArrayStr(vs) => Ok(Box::new(vs.to_owned())),
+        ParameterValue::Interval(v) => Ok(Box::new(Interval(*v))),
         ParameterValue::DbNull => Ok(Box::new(PgNull)),
     }
 }
@@ -442,6 +443,13 @@ fn convert_entry(row: &Row, index: usize) -> anyhow::Result<DbValue> {
                 None => DbValue::DbNull,
             }
         }
+        &Type::INTERVAL => {
+            let value: Option<Interval> = row.try_get(index)?;
+            match value {
+                Some(v) => DbValue::Interval(v.0),
+                None => DbValue::DbNull,
+            }
+        }
         t => {
             tracing::debug!(
                 "Couldn't convert Postgres type {} in column {}",
@@ -544,6 +552,83 @@ impl std::fmt::Debug for PgNull {
     }
 }
 
+#[derive(Debug)]
+struct Interval(v4::Interval);
+
+impl ToSql for Interval {
+    tokio_postgres::types::to_sql_checked!();
+
+    fn to_sql(
+        &self,
+        _ty: &Type,
+        out: &mut tokio_postgres::types::private::BytesMut,
+    ) -> Result<tokio_postgres::types::IsNull, Box<dyn std::error::Error + Sync + Send>>
+    where
+        Self: Sized,
+    {
+        use bytes::BufMut;
+
+        out.put_i64(self.0.micros);
+        out.put_i32(self.0.days);
+        out.put_i32(self.0.months);
+
+        Ok(tokio_postgres::types::IsNull::No)
+    }
+
+    fn accepts(ty: &Type) -> bool
+    where
+        Self: Sized,
+    {
+        matches!(ty, &Type::INTERVAL)
+    }
+}
+
+impl FromSql<'_> for Interval {
+    fn from_sql(
+        _ty: &Type,
+        raw: &'_ [u8],
+    ) -> std::result::Result<Self, Box<dyn std::error::Error + Sync + Send>> {
+        const EXPECTED_LEN: usize = size_of::<i64>() + size_of::<i32>() + size_of::<i32>();
+
+        if raw.len() != EXPECTED_LEN {
+            return Err(Box::new(IntervalLengthError));
+        }
+
+        let (micro_bytes, rest) = raw.split_at(size_of::<i64>());
+        let (day_bytes, rest) = rest.split_at(size_of::<i32>());
+        let month_bytes = rest;
+        let months = i32::from_be_bytes(month_bytes.try_into().unwrap());
+        let days = i32::from_be_bytes(day_bytes.try_into().unwrap());
+        let micros = i64::from_be_bytes(micro_bytes.try_into().unwrap());
+
+        Ok(Self(v4::Interval {
+            micros,
+            days,
+            months,
+        }))
+    }
+
+    fn accepts(ty: &Type) -> bool {
+        matches!(ty, &Type::INTERVAL)
+    }
+}
+
+struct IntervalLengthError;
+
+impl std::error::Error for IntervalLengthError {}
+
+impl std::fmt::Display for IntervalLengthError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("unexpected binary format for Postgres INTERVAL")
+    }
+}
+
+impl std::fmt::Debug for IntervalLengthError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        std::fmt::Display::fmt(self, f)
+    }
+}
+
 /// Workaround for moka returning Arc<Error> which, although
 /// necessary for concurrency, does not play well with others.
 struct ArcError(std::sync::Arc<anyhow::Error>);

crates/factor-outbound-pg/src/client.rs

@@ -125,6 +125,7 @@ mod rdbms_types {
                 pg4::DbValue::ArrayInt32(_) => pg3::DbValue::Unsupported,
                 pg4::DbValue::ArrayInt64(_) => pg3::DbValue::Unsupported,
                 pg4::DbValue::ArrayStr(_) => pg3::DbValue::Unsupported,
+                pg4::DbValue::Interval(_) => pg3::DbValue::Unsupported,
                 pg4::DbValue::DbNull => pg3::DbValue::DbNull,
                 pg4::DbValue::Unsupported => pg3::DbValue::Unsupported,
             }
@@ -191,6 +192,7 @@ mod rdbms_types {
                 pg4::DbDataType::ArrayInt32 => pg3::DbDataType::Other,
                 pg4::DbDataType::ArrayInt64 => pg3::DbDataType::Other,
                 pg4::DbDataType::ArrayStr => pg3::DbDataType::Other,
+                pg4::DbDataType::Interval => pg3::DbDataType::Other,
                 pg4::DbDataType::Other => pg3::DbDataType::Other,
             }
         }

crates/world/src/conversions.rs

@@ -33,6 +33,7 @@ interface postgres {
       array-int32,
       array-int64,
       array-str,
+      interval,
       other,
   }
 
@@ -62,6 +63,7 @@ interface postgres {
       array-int32(list<option<s32>>),
       array-int64(list<option<s64>>),
       array-str(list<option<string>>),
+      interval(interval),
       db-null,
       unsupported,
   }
@@ -92,9 +94,16 @@ interface postgres {
       array-int32(list<option<s32>>),
       array-int64(list<option<s64>>),
       array-str(list<option<string>>),
+      interval(interval),
       db-null,
   }
 
+  record interval {
+    micros: s64,
+    days: s32,
+    months: s32,
+  }
+
   /// A database column
   record column {
       name: string,