use bytes::{Buf, BytesMut};
use tokio_postgres::types::{FromSql, IsNull, ToSql, Type};
use tokio_postgres::Row as PgRow;
use oxisql_core::{ArrayElementType, Row, Value};
use crate::error::PgError;
struct RawNumeric(Vec<u8>);
impl<'a> FromSql<'a> for RawNumeric {
fn from_sql(
_ty: &Type,
raw: &'a [u8],
) -> Result<Self, Box<dyn std::error::Error + Sync + Send>> {
Ok(RawNumeric(raw.to_vec()))
}
fn accepts(ty: &Type) -> bool {
*ty == Type::NUMERIC
}
}
fn decode_pg_numeric(raw: &[u8]) -> Result<String, PgError> {
if raw.len() < 8 {
return Err(PgError::TypeConversion(format!(
"NUMERIC binary too short: {} bytes",
raw.len()
)));
}
let mut buf = raw;
let ndigits = buf.get_u16() as usize;
let weight = buf.get_i16(); let sign = buf.get_u16();
let dscale = buf.get_u16() as usize;
const SIGN_NAN: u16 = 0xC000;
const SIGN_NEG: u16 = 0x4000;
if sign == SIGN_NAN {
return Ok("NaN".to_string());
}
let mut digits: Vec<u16> = Vec::with_capacity(ndigits);
for _ in 0..ndigits {
if buf.remaining() < 2 {
return Err(PgError::TypeConversion(
"NUMERIC binary truncated while reading digit groups".to_string(),
));
}
digits.push(buf.get_u16());
}
let mut integer_part = String::new();
let mut fraction_part = String::new();
let int_groups = (weight + 1).max(0) as usize;
let int_digit_iter = (0..int_groups).map(|i| if i < ndigits { digits[i] } else { 0 });
for (i, digit) in int_digit_iter.enumerate() {
if i == 0 {
integer_part.push_str(&digit.to_string());
} else {
integer_part.push_str(&format!("{digit:04}"));
}
}
if integer_part.is_empty() {
integer_part.push('0');
}
if dscale > 0 {
let mut remaining_scale = dscale;
let frac_groups_needed = dscale.div_ceil(4);
for i in 0..frac_groups_needed {
let digit_idx = int_groups + i;
let digit = if digit_idx < ndigits {
digits[digit_idx]
} else {
0
};
let group_str = format!("{digit:04}");
let take = remaining_scale.min(4);
fraction_part.push_str(&group_str[..take]);
remaining_scale = remaining_scale.saturating_sub(4);
}
}
let mut result = if sign == SIGN_NEG {
format!("-{integer_part}")
} else {
integer_part
};
if !fraction_part.is_empty() {
result.push('.');
result.push_str(&fraction_part);
}
Ok(result)
}
struct RawInterval(Vec<u8>);
impl<'a> FromSql<'a> for RawInterval {
fn from_sql(
_ty: &Type,
raw: &'a [u8],
) -> Result<Self, Box<dyn std::error::Error + Sync + Send>> {
Ok(RawInterval(raw.to_vec()))
}
fn accepts(ty: &Type) -> bool {
*ty == Type::INTERVAL
}
}
fn decode_pg_interval(raw: &[u8]) -> Result<String, PgError> {
if raw.len() < 16 {
return Err(PgError::TypeConversion(format!(
"INTERVAL binary too short: {} bytes",
raw.len()
)));
}
let mut buf = raw;
let micros = buf.get_i64();
let days = buf.get_i32();
let months = buf.get_i32();
let years = months / 12;
let rem_months = months % 12;
let hours = micros / 3_600_000_000;
let rem_micros = micros % 3_600_000_000;
let minutes = rem_micros / 60_000_000;
let rem_micros2 = rem_micros % 60_000_000;
let secs = rem_micros2 / 1_000_000;
let frac_micros = (rem_micros2 % 1_000_000).unsigned_abs();
let mut parts: Vec<String> = Vec::new();
if years != 0 {
parts.push(format!("{years}Y"));
}
if rem_months != 0 {
parts.push(format!("{rem_months}M"));
}
if days != 0 {
parts.push(format!("{days}D"));
}
let has_time = hours != 0 || minutes != 0 || secs != 0 || frac_micros != 0;
if has_time {
if frac_micros > 0 {
parts.push(format!(
"{hours:02}:{minutes:02}:{secs:02}.{frac_micros:06}"
));
} else {
parts.push(format!("{hours:02}:{minutes:02}:{secs:02}"));
}
}
if parts.is_empty() {
Ok("00:00:00".to_string())
} else {
Ok(parts.join(" "))
}
}
#[derive(Debug)]
pub enum OwnedParam {
Null,
Bool(bool),
I64(i64),
F64(f64),
Text(String),
Blob(Vec<u8>),
}
impl ToSql for OwnedParam {
fn to_sql(
&self,
ty: &Type,
out: &mut BytesMut,
) -> Result<IsNull, Box<dyn std::error::Error + Sync + Send>> {
match self {
OwnedParam::Null => Ok(IsNull::Yes),
OwnedParam::Bool(v) => v.to_sql(ty, out),
OwnedParam::I64(v) => {
if *ty == Type::INT4 {
let v4 = i32::try_from(*v)
.map_err(|e| Box::new(e) as Box<dyn std::error::Error + Sync + Send>)?;
v4.to_sql(ty, out)
} else if *ty == Type::INT2 {
let v2 = i16::try_from(*v)
.map_err(|e| Box::new(e) as Box<dyn std::error::Error + Sync + Send>)?;
v2.to_sql(ty, out)
} else {
v.to_sql(ty, out)
}
}
OwnedParam::F64(v) => {
if *ty == Type::FLOAT4 {
#[allow(clippy::cast_possible_truncation)]
let v4 = *v as f32;
v4.to_sql(ty, out)
} else {
v.to_sql(ty, out)
}
}
OwnedParam::Text(v) => v.to_sql(ty, out),
OwnedParam::Blob(v) => v.to_sql(ty, out),
}
}
fn accepts(_ty: &Type) -> bool
where
Self: Sized,
{
true
}
tokio_postgres::types::to_sql_checked!();
}
pub fn value_to_param(v: &Value) -> OwnedParam {
match v {
Value::Null => OwnedParam::Null,
Value::Bool(b) => OwnedParam::Bool(*b),
Value::I64(n) => OwnedParam::I64(*n),
Value::F64(f) => OwnedParam::F64(*f),
Value::Text(s) => OwnedParam::Text(s.clone()),
Value::Blob(b) => OwnedParam::Blob(b.clone()),
Value::Timestamp(us) => {
let secs = us / 1_000_000;
let frac = (us % 1_000_000).unsigned_abs();
OwnedParam::Text(format!("{secs}.{frac:06}"))
}
Value::Date(days) => OwnedParam::Text(format!("{days}")),
Value::Time(us) => {
let total_secs = us / 1_000_000;
let hours = total_secs / 3600;
let mins = (total_secs % 3600) / 60;
let secs = total_secs % 60;
let frac = (us % 1_000_000).unsigned_abs();
if frac == 0 {
OwnedParam::Text(format!("{hours:02}:{mins:02}:{secs:02}"))
} else {
OwnedParam::Text(format!("{hours:02}:{mins:02}:{secs:02}.{frac:06}"))
}
}
Value::Uuid(u) => OwnedParam::Text(format!("{}", Value::Uuid(*u))),
Value::Json(s) => OwnedParam::Text(s.clone()),
Value::Decimal(s) => OwnedParam::Text(s.clone()),
Value::Array(vals) => {
let items: Vec<String> = vals.iter().map(|v| format!("{v}")).collect();
OwnedParam::Text(format!("{{{}}}", items.join(",")))
}
Value::TypedArray { values, .. } => {
let items: Vec<String> = values.iter().map(|v| format!("{v}")).collect();
OwnedParam::Text(format!("{{{}}}", items.join(",")))
}
}
}
pub fn pg_row_to_row(pg_row: PgRow) -> Result<Row, PgError> {
let cols = pg_row.columns();
let mut names = Vec::with_capacity(cols.len());
let mut values = Vec::with_capacity(cols.len());
for (i, col) in cols.iter().enumerate() {
names.push(col.name().to_string());
let value = extract_value(&pg_row, i, col.type_())?;
values.push(value);
}
Ok(Row::new(names, values))
}
fn extract_value(row: &PgRow, idx: usize, ty: &Type) -> Result<Value, PgError> {
match *ty {
Type::BOOL => {
let v: Option<bool> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(Value::Bool).unwrap_or(Value::Null))
}
Type::INT2 => {
let v: Option<i16> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|n| Value::I64(i64::from(n))).unwrap_or(Value::Null))
}
Type::INT4 => {
let v: Option<i32> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|n| Value::I64(i64::from(n))).unwrap_or(Value::Null))
}
Type::INT8 => {
let v: Option<i64> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(Value::I64).unwrap_or(Value::Null))
}
Type::FLOAT4 => {
let v: Option<f32> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|f| Value::F64(f64::from(f))).unwrap_or(Value::Null))
}
Type::FLOAT8 => {
let v: Option<f64> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(Value::F64).unwrap_or(Value::Null))
}
Type::BYTEA => {
let v: Option<Vec<u8>> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(Value::Blob).unwrap_or(Value::Null))
}
Type::JSON | Type::JSONB => {
let v: Option<String> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(Value::Json).unwrap_or(Value::Null))
}
Type::DATE => {
let v: Option<time::Date> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|d| {
const UNIX_EPOCH_JDN: i32 = 2_440_588; Value::Date(d.to_julian_day() - UNIX_EPOCH_JDN)
})
.unwrap_or(Value::Null))
}
Type::TIMESTAMP => {
let v: Option<time::PrimitiveDateTime> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|dt| {
let us = dt.assume_utc().unix_timestamp_nanos() / 1_000;
#[allow(clippy::cast_possible_truncation)]
Value::Timestamp(us as i64)
})
.unwrap_or(Value::Null))
}
Type::TIMESTAMPTZ => {
let v: Option<time::OffsetDateTime> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|dt| {
let us = dt.unix_timestamp_nanos() / 1_000;
#[allow(clippy::cast_possible_truncation)]
Value::Timestamp(us as i64)
})
.unwrap_or(Value::Null))
}
Type::TIME => {
let v: Option<time::Time> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|t| {
let (h, m, s, ns) = t.as_hms_nano();
let us = i64::from(h) * 3_600_000_000
+ i64::from(m) * 60_000_000
+ i64::from(s) * 1_000_000
+ i64::from(ns) / 1_000;
Value::Time(us)
})
.unwrap_or(Value::Null))
}
Type::UUID => {
let v: Option<uuid::Uuid> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|u| Value::Uuid(u.as_u128())).unwrap_or(Value::Null))
}
Type::NUMERIC => {
let v: Option<RawNumeric> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
match v {
None => Ok(Value::Null),
Some(raw) => {
let s = decode_pg_numeric(&raw.0)?;
Ok(Value::Decimal(s))
}
}
}
Type::INTERVAL => {
let v: Option<RawInterval> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
match v {
None => Ok(Value::Null),
Some(raw) => {
let s = decode_pg_interval(&raw.0)?;
Ok(Value::Text(s))
}
}
}
Type::BOOL_ARRAY => {
let v: Option<Vec<Option<bool>>> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|items| Value::TypedArray {
element_type: ArrayElementType::Bool,
values: items
.into_iter()
.map(|opt| opt.map(Value::Bool).unwrap_or(Value::Null))
.collect(),
})
.unwrap_or(Value::Null))
}
Type::INT2_ARRAY => {
let v: Option<Vec<Option<i16>>> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|items| Value::TypedArray {
element_type: ArrayElementType::Int2,
values: items
.into_iter()
.map(|opt| opt.map(|n| Value::I64(i64::from(n))).unwrap_or(Value::Null))
.collect(),
})
.unwrap_or(Value::Null))
}
Type::INT4_ARRAY => {
let v: Option<Vec<Option<i32>>> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|items| Value::TypedArray {
element_type: ArrayElementType::Int4,
values: items
.into_iter()
.map(|opt| opt.map(|n| Value::I64(i64::from(n))).unwrap_or(Value::Null))
.collect(),
})
.unwrap_or(Value::Null))
}
Type::INT8_ARRAY => {
let v: Option<Vec<Option<i64>>> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|items| Value::TypedArray {
element_type: ArrayElementType::Int8,
values: items
.into_iter()
.map(|opt| opt.map(Value::I64).unwrap_or(Value::Null))
.collect(),
})
.unwrap_or(Value::Null))
}
Type::FLOAT4_ARRAY => {
let v: Option<Vec<Option<f32>>> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|items| Value::TypedArray {
element_type: ArrayElementType::Float4,
values: items
.into_iter()
.map(|opt| opt.map(|f| Value::F64(f64::from(f))).unwrap_or(Value::Null))
.collect(),
})
.unwrap_or(Value::Null))
}
Type::FLOAT8_ARRAY => {
let v: Option<Vec<Option<f64>>> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|items| Value::TypedArray {
element_type: ArrayElementType::Float8,
values: items
.into_iter()
.map(|opt| opt.map(Value::F64).unwrap_or(Value::Null))
.collect(),
})
.unwrap_or(Value::Null))
}
Type::TEXT_ARRAY | Type::VARCHAR_ARRAY => {
let v: Option<Vec<Option<String>>> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|items| Value::TypedArray {
element_type: ArrayElementType::Text,
values: items
.into_iter()
.map(|opt| opt.map(Value::Text).unwrap_or(Value::Null))
.collect(),
})
.unwrap_or(Value::Null))
}
Type::TIMESTAMPTZ_ARRAY => {
let v: Option<Vec<Option<time::OffsetDateTime>>> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|items| Value::TypedArray {
element_type: ArrayElementType::TimestampTz,
values: items
.into_iter()
.map(|opt| {
opt.map(|dt| {
let us = dt.unix_timestamp_nanos() / 1_000;
#[allow(clippy::cast_possible_truncation)]
Value::Timestamp(us as i64)
})
.unwrap_or(Value::Null)
})
.collect(),
})
.unwrap_or(Value::Null))
}
Type::TIMESTAMP_ARRAY => {
let v: Option<Vec<Option<time::PrimitiveDateTime>>> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|items| Value::TypedArray {
element_type: ArrayElementType::Timestamp,
values: items
.into_iter()
.map(|opt| {
opt.map(|dt| {
let us = dt.assume_utc().unix_timestamp_nanos() / 1_000;
#[allow(clippy::cast_possible_truncation)]
Value::Timestamp(us as i64)
})
.unwrap_or(Value::Null)
})
.collect(),
})
.unwrap_or(Value::Null))
}
Type::DATE_ARRAY => {
let v: Option<Vec<Option<time::Date>>> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|items| Value::TypedArray {
element_type: ArrayElementType::Date,
values: items
.into_iter()
.map(|opt| {
opt.map(|d| {
const UNIX_EPOCH_JDN: i32 = 2_440_588;
Value::Date(d.to_julian_day() - UNIX_EPOCH_JDN)
})
.unwrap_or(Value::Null)
})
.collect(),
})
.unwrap_or(Value::Null))
}
Type::TIME_ARRAY => {
let v: Option<Vec<Option<time::Time>>> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|items| Value::TypedArray {
element_type: ArrayElementType::Time,
values: items
.into_iter()
.map(|opt| {
opt.map(|t| {
let (h, m, s, ns) = t.as_hms_nano();
let us = i64::from(h) * 3_600_000_000
+ i64::from(m) * 60_000_000
+ i64::from(s) * 1_000_000
+ i64::from(ns) / 1_000;
Value::Time(us)
})
.unwrap_or(Value::Null)
})
.collect(),
})
.unwrap_or(Value::Null))
}
Type::UUID_ARRAY => {
let v: Option<Vec<Option<uuid::Uuid>>> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|items| Value::TypedArray {
element_type: ArrayElementType::Uuid,
values: items
.into_iter()
.map(|opt| opt.map(|u| Value::Uuid(u.as_u128())).unwrap_or(Value::Null))
.collect(),
})
.unwrap_or(Value::Null))
}
Type::BYTEA_ARRAY => {
let v: Option<Vec<Option<Vec<u8>>>> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|items| Value::TypedArray {
element_type: ArrayElementType::Bytea,
values: items
.into_iter()
.map(|opt| opt.map(Value::Blob).unwrap_or(Value::Null))
.collect(),
})
.unwrap_or(Value::Null))
}
Type::JSON_ARRAY => {
let v: Option<Vec<Option<String>>> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|items| Value::TypedArray {
element_type: ArrayElementType::Json,
values: items
.into_iter()
.map(|opt| opt.map(Value::Json).unwrap_or(Value::Null))
.collect(),
})
.unwrap_or(Value::Null))
}
Type::JSONB_ARRAY => {
let v: Option<Vec<Option<String>>> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(|items| Value::TypedArray {
element_type: ArrayElementType::Jsonb,
values: items
.into_iter()
.map(|opt| opt.map(Value::Json).unwrap_or(Value::Null))
.collect(),
})
.unwrap_or(Value::Null))
}
ref other => {
if <String as FromSql>::accepts(other) {
let v: Option<String> = row
.try_get(idx)
.map_err(|e| PgError::TypeConversion(e.to_string()))?;
Ok(v.map(Value::Text).unwrap_or(Value::Null))
} else {
Ok(Value::Text(format!("<opaque:{}>", other.name())))
}
}
}
}