1use bytes::{Buf, BytesMut};
28use tokio_postgres::types::{FromSql, IsNull, ToSql, Type};
29use tokio_postgres::Row as PgRow;
30
31use oxisql_core::{ArrayElementType, Row, Value};
32
33use crate::error::PgError;
34
35struct RawNumeric(Vec<u8>);
47
48impl<'a> FromSql<'a> for RawNumeric {
49 fn from_sql(
50 _ty: &Type,
51 raw: &'a [u8],
52 ) -> Result<Self, Box<dyn std::error::Error + Sync + Send>> {
53 Ok(RawNumeric(raw.to_vec()))
54 }
55
56 fn accepts(ty: &Type) -> bool {
57 *ty == Type::NUMERIC
58 }
59}
60
61fn decode_pg_numeric(raw: &[u8]) -> Result<String, PgError> {
66 if raw.len() < 8 {
67 return Err(PgError::TypeConversion(format!(
68 "NUMERIC binary too short: {} bytes",
69 raw.len()
70 )));
71 }
72
73 let mut buf = raw;
74
75 let ndigits = buf.get_u16() as usize;
76 let weight = buf.get_i16(); let sign = buf.get_u16();
78 let dscale = buf.get_u16() as usize;
79
80 const SIGN_NAN: u16 = 0xC000;
81 const SIGN_NEG: u16 = 0x4000;
82
83 if sign == SIGN_NAN {
84 return Ok("NaN".to_string());
85 }
86
87 let mut digits: Vec<u16> = Vec::with_capacity(ndigits);
89 for _ in 0..ndigits {
90 if buf.remaining() < 2 {
91 return Err(PgError::TypeConversion(
92 "NUMERIC binary truncated while reading digit groups".to_string(),
93 ));
94 }
95 digits.push(buf.get_u16());
96 }
97
98 let mut integer_part = String::new();
108 let mut fraction_part = String::new();
109
110 let int_groups = (weight + 1).max(0) as usize;
112
113 let int_digit_iter = (0..int_groups).map(|i| if i < ndigits { digits[i] } else { 0 });
116 for (i, digit) in int_digit_iter.enumerate() {
117 if i == 0 {
118 integer_part.push_str(&digit.to_string());
120 } else {
121 integer_part.push_str(&format!("{digit:04}"));
123 }
124 }
125
126 if integer_part.is_empty() {
127 integer_part.push('0');
128 }
129
130 if dscale > 0 {
133 let mut remaining_scale = dscale;
134 let frac_groups_needed = dscale.div_ceil(4);
135
136 for i in 0..frac_groups_needed {
137 let digit_idx = int_groups + i;
138 let digit = if digit_idx < ndigits {
139 digits[digit_idx]
140 } else {
141 0
142 };
143 let group_str = format!("{digit:04}");
144 let take = remaining_scale.min(4);
146 fraction_part.push_str(&group_str[..take]);
147 remaining_scale = remaining_scale.saturating_sub(4);
148 }
149 }
150
151 let mut result = if sign == SIGN_NEG {
152 format!("-{integer_part}")
153 } else {
154 integer_part
155 };
156
157 if !fraction_part.is_empty() {
158 result.push('.');
159 result.push_str(&fraction_part);
160 }
161
162 Ok(result)
163}
164
165struct RawInterval(Vec<u8>);
174
175impl<'a> FromSql<'a> for RawInterval {
176 fn from_sql(
177 _ty: &Type,
178 raw: &'a [u8],
179 ) -> Result<Self, Box<dyn std::error::Error + Sync + Send>> {
180 Ok(RawInterval(raw.to_vec()))
181 }
182
183 fn accepts(ty: &Type) -> bool {
184 *ty == Type::INTERVAL
185 }
186}
187
188fn decode_pg_interval(raw: &[u8]) -> Result<String, PgError> {
192 if raw.len() < 16 {
193 return Err(PgError::TypeConversion(format!(
194 "INTERVAL binary too short: {} bytes",
195 raw.len()
196 )));
197 }
198 let mut buf = raw;
199 let micros = buf.get_i64();
200 let days = buf.get_i32();
201 let months = buf.get_i32();
202
203 let years = months / 12;
204 let rem_months = months % 12;
205 let hours = micros / 3_600_000_000;
206 let rem_micros = micros % 3_600_000_000;
207 let minutes = rem_micros / 60_000_000;
208 let rem_micros2 = rem_micros % 60_000_000;
209 let secs = rem_micros2 / 1_000_000;
210 let frac_micros = (rem_micros2 % 1_000_000).unsigned_abs();
211
212 let mut parts: Vec<String> = Vec::new();
213 if years != 0 {
214 parts.push(format!("{years}Y"));
215 }
216 if rem_months != 0 {
217 parts.push(format!("{rem_months}M"));
218 }
219 if days != 0 {
220 parts.push(format!("{days}D"));
221 }
222
223 let has_time = hours != 0 || minutes != 0 || secs != 0 || frac_micros != 0;
224 if has_time {
225 if frac_micros > 0 {
226 parts.push(format!(
227 "{hours:02}:{minutes:02}:{secs:02}.{frac_micros:06}"
228 ));
229 } else {
230 parts.push(format!("{hours:02}:{minutes:02}:{secs:02}"));
231 }
232 }
233
234 if parts.is_empty() {
235 Ok("00:00:00".to_string())
236 } else {
237 Ok(parts.join(" "))
238 }
239}
240
241#[derive(Debug)]
249pub enum OwnedParam {
250 Null,
252 Bool(bool),
254 I64(i64),
256 F64(f64),
258 Text(String),
260 Blob(Vec<u8>),
262}
263
264impl ToSql for OwnedParam {
265 fn to_sql(
266 &self,
267 ty: &Type,
268 out: &mut BytesMut,
269 ) -> Result<IsNull, Box<dyn std::error::Error + Sync + Send>> {
270 match self {
271 OwnedParam::Null => Ok(IsNull::Yes),
272 OwnedParam::Bool(v) => v.to_sql(ty, out),
273 OwnedParam::I64(v) => {
274 if *ty == Type::INT4 {
275 let v4 = i32::try_from(*v)
276 .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Sync + Send>)?;
277 v4.to_sql(ty, out)
278 } else if *ty == Type::INT2 {
279 let v2 = i16::try_from(*v)
280 .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Sync + Send>)?;
281 v2.to_sql(ty, out)
282 } else {
283 v.to_sql(ty, out)
284 }
285 }
286 OwnedParam::F64(v) => {
287 if *ty == Type::FLOAT4 {
288 #[allow(clippy::cast_possible_truncation)]
289 let v4 = *v as f32;
290 v4.to_sql(ty, out)
291 } else {
292 v.to_sql(ty, out)
293 }
294 }
295 OwnedParam::Text(v) => v.to_sql(ty, out),
296 OwnedParam::Blob(v) => v.to_sql(ty, out),
297 }
298 }
299
300 fn accepts(_ty: &Type) -> bool
301 where
302 Self: Sized,
303 {
304 true
307 }
308
309 tokio_postgres::types::to_sql_checked!();
310}
311
312pub fn value_to_param(v: &Value) -> OwnedParam {
318 match v {
319 Value::Null => OwnedParam::Null,
320 Value::Bool(b) => OwnedParam::Bool(*b),
321 Value::I64(n) => OwnedParam::I64(*n),
322 Value::F64(f) => OwnedParam::F64(*f),
323 Value::Text(s) => OwnedParam::Text(s.clone()),
324 Value::Blob(b) => OwnedParam::Blob(b.clone()),
325 Value::Timestamp(us) => {
327 let secs = us / 1_000_000;
328 let frac = (us % 1_000_000).unsigned_abs();
329 OwnedParam::Text(format!("{secs}.{frac:06}"))
330 }
331 Value::Date(days) => OwnedParam::Text(format!("{days}")),
332 Value::Time(us) => {
333 let total_secs = us / 1_000_000;
334 let hours = total_secs / 3600;
335 let mins = (total_secs % 3600) / 60;
336 let secs = total_secs % 60;
337 let frac = (us % 1_000_000).unsigned_abs();
338 if frac == 0 {
339 OwnedParam::Text(format!("{hours:02}:{mins:02}:{secs:02}"))
340 } else {
341 OwnedParam::Text(format!("{hours:02}:{mins:02}:{secs:02}.{frac:06}"))
342 }
343 }
344 Value::Uuid(u) => OwnedParam::Text(format!("{}", Value::Uuid(*u))),
345 Value::Json(s) => OwnedParam::Text(s.clone()),
346 Value::Decimal(s) => OwnedParam::Text(s.clone()),
347 Value::Array(vals) => {
348 let items: Vec<String> = vals.iter().map(|v| format!("{v}")).collect();
350 OwnedParam::Text(format!("{{{}}}", items.join(",")))
351 }
352 Value::TypedArray { values, .. } => {
353 let items: Vec<String> = values.iter().map(|v| format!("{v}")).collect();
355 OwnedParam::Text(format!("{{{}}}", items.join(",")))
356 }
357 }
358}
359
360pub fn pg_row_to_row(pg_row: PgRow) -> Result<Row, PgError> {
368 let cols = pg_row.columns();
369 let mut names = Vec::with_capacity(cols.len());
370 let mut values = Vec::with_capacity(cols.len());
371
372 for (i, col) in cols.iter().enumerate() {
373 names.push(col.name().to_string());
374 let value = extract_value(&pg_row, i, col.type_())?;
375 values.push(value);
376 }
377
378 Ok(Row::new(names, values))
379}
380
381fn extract_value(row: &PgRow, idx: usize, ty: &Type) -> Result<Value, PgError> {
382 match *ty {
383 Type::BOOL => {
384 let v: Option<bool> = row
385 .try_get(idx)
386 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
387 Ok(v.map(Value::Bool).unwrap_or(Value::Null))
388 }
389 Type::INT2 => {
390 let v: Option<i16> = row
391 .try_get(idx)
392 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
393 Ok(v.map(|n| Value::I64(i64::from(n))).unwrap_or(Value::Null))
394 }
395 Type::INT4 => {
396 let v: Option<i32> = row
397 .try_get(idx)
398 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
399 Ok(v.map(|n| Value::I64(i64::from(n))).unwrap_or(Value::Null))
400 }
401 Type::INT8 => {
402 let v: Option<i64> = row
403 .try_get(idx)
404 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
405 Ok(v.map(Value::I64).unwrap_or(Value::Null))
406 }
407 Type::FLOAT4 => {
408 let v: Option<f32> = row
409 .try_get(idx)
410 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
411 Ok(v.map(|f| Value::F64(f64::from(f))).unwrap_or(Value::Null))
412 }
413 Type::FLOAT8 => {
414 let v: Option<f64> = row
415 .try_get(idx)
416 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
417 Ok(v.map(Value::F64).unwrap_or(Value::Null))
418 }
419 Type::BYTEA => {
420 let v: Option<Vec<u8>> = row
421 .try_get(idx)
422 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
423 Ok(v.map(Value::Blob).unwrap_or(Value::Null))
424 }
425 Type::JSON | Type::JSONB => {
427 let v: Option<String> = row
428 .try_get(idx)
429 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
430 Ok(v.map(Value::Json).unwrap_or(Value::Null))
431 }
432 Type::DATE => {
440 let v: Option<time::Date> = row
441 .try_get(idx)
442 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
443 Ok(v.map(|d| {
444 const UNIX_EPOCH_JDN: i32 = 2_440_588; Value::Date(d.to_julian_day() - UNIX_EPOCH_JDN)
446 })
447 .unwrap_or(Value::Null))
448 }
449 Type::TIMESTAMP => {
451 let v: Option<time::PrimitiveDateTime> = row
452 .try_get(idx)
453 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
454 Ok(v.map(|dt| {
455 let us = dt.assume_utc().unix_timestamp_nanos() / 1_000;
457 #[allow(clippy::cast_possible_truncation)]
458 Value::Timestamp(us as i64)
459 })
460 .unwrap_or(Value::Null))
461 }
462 Type::TIMESTAMPTZ => {
464 let v: Option<time::OffsetDateTime> = row
465 .try_get(idx)
466 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
467 Ok(v.map(|dt| {
468 let us = dt.unix_timestamp_nanos() / 1_000;
469 #[allow(clippy::cast_possible_truncation)]
470 Value::Timestamp(us as i64)
471 })
472 .unwrap_or(Value::Null))
473 }
474 Type::TIME => {
476 let v: Option<time::Time> = row
477 .try_get(idx)
478 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
479 Ok(v.map(|t| {
480 let (h, m, s, ns) = t.as_hms_nano();
481 let us = i64::from(h) * 3_600_000_000
482 + i64::from(m) * 60_000_000
483 + i64::from(s) * 1_000_000
484 + i64::from(ns) / 1_000;
485 Value::Time(us)
486 })
487 .unwrap_or(Value::Null))
488 }
489 Type::UUID => {
491 let v: Option<uuid::Uuid> = row
492 .try_get(idx)
493 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
494 Ok(v.map(|u| Value::Uuid(u.as_u128())).unwrap_or(Value::Null))
495 }
496 Type::NUMERIC => {
498 let v: Option<RawNumeric> = row
499 .try_get(idx)
500 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
501 match v {
502 None => Ok(Value::Null),
503 Some(raw) => {
504 let s = decode_pg_numeric(&raw.0)?;
505 Ok(Value::Decimal(s))
506 }
507 }
508 }
509 Type::INTERVAL => {
511 let v: Option<RawInterval> = row
512 .try_get(idx)
513 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
514 match v {
515 None => Ok(Value::Null),
516 Some(raw) => {
517 let s = decode_pg_interval(&raw.0)?;
518 Ok(Value::Text(s))
519 }
520 }
521 }
522 Type::BOOL_ARRAY => {
525 let v: Option<Vec<Option<bool>>> = row
526 .try_get(idx)
527 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
528 Ok(v.map(|items| Value::TypedArray {
529 element_type: ArrayElementType::Bool,
530 values: items
531 .into_iter()
532 .map(|opt| opt.map(Value::Bool).unwrap_or(Value::Null))
533 .collect(),
534 })
535 .unwrap_or(Value::Null))
536 }
537 Type::INT2_ARRAY => {
538 let v: Option<Vec<Option<i16>>> = row
539 .try_get(idx)
540 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
541 Ok(v.map(|items| Value::TypedArray {
542 element_type: ArrayElementType::Int2,
543 values: items
544 .into_iter()
545 .map(|opt| opt.map(|n| Value::I64(i64::from(n))).unwrap_or(Value::Null))
546 .collect(),
547 })
548 .unwrap_or(Value::Null))
549 }
550 Type::INT4_ARRAY => {
551 let v: Option<Vec<Option<i32>>> = row
552 .try_get(idx)
553 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
554 Ok(v.map(|items| Value::TypedArray {
555 element_type: ArrayElementType::Int4,
556 values: items
557 .into_iter()
558 .map(|opt| opt.map(|n| Value::I64(i64::from(n))).unwrap_or(Value::Null))
559 .collect(),
560 })
561 .unwrap_or(Value::Null))
562 }
563 Type::INT8_ARRAY => {
564 let v: Option<Vec<Option<i64>>> = row
565 .try_get(idx)
566 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
567 Ok(v.map(|items| Value::TypedArray {
568 element_type: ArrayElementType::Int8,
569 values: items
570 .into_iter()
571 .map(|opt| opt.map(Value::I64).unwrap_or(Value::Null))
572 .collect(),
573 })
574 .unwrap_or(Value::Null))
575 }
576 Type::FLOAT4_ARRAY => {
577 let v: Option<Vec<Option<f32>>> = row
578 .try_get(idx)
579 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
580 Ok(v.map(|items| Value::TypedArray {
581 element_type: ArrayElementType::Float4,
582 values: items
583 .into_iter()
584 .map(|opt| opt.map(|f| Value::F64(f64::from(f))).unwrap_or(Value::Null))
585 .collect(),
586 })
587 .unwrap_or(Value::Null))
588 }
589 Type::FLOAT8_ARRAY => {
590 let v: Option<Vec<Option<f64>>> = row
591 .try_get(idx)
592 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
593 Ok(v.map(|items| Value::TypedArray {
594 element_type: ArrayElementType::Float8,
595 values: items
596 .into_iter()
597 .map(|opt| opt.map(Value::F64).unwrap_or(Value::Null))
598 .collect(),
599 })
600 .unwrap_or(Value::Null))
601 }
602 Type::TEXT_ARRAY | Type::VARCHAR_ARRAY => {
603 let v: Option<Vec<Option<String>>> = row
604 .try_get(idx)
605 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
606 Ok(v.map(|items| Value::TypedArray {
607 element_type: ArrayElementType::Text,
608 values: items
609 .into_iter()
610 .map(|opt| opt.map(Value::Text).unwrap_or(Value::Null))
611 .collect(),
612 })
613 .unwrap_or(Value::Null))
614 }
615 Type::TIMESTAMPTZ_ARRAY => {
617 let v: Option<Vec<Option<time::OffsetDateTime>>> = row
618 .try_get(idx)
619 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
620 Ok(v.map(|items| Value::TypedArray {
621 element_type: ArrayElementType::TimestampTz,
622 values: items
623 .into_iter()
624 .map(|opt| {
625 opt.map(|dt| {
626 let us = dt.unix_timestamp_nanos() / 1_000;
627 #[allow(clippy::cast_possible_truncation)]
628 Value::Timestamp(us as i64)
629 })
630 .unwrap_or(Value::Null)
631 })
632 .collect(),
633 })
634 .unwrap_or(Value::Null))
635 }
636 Type::TIMESTAMP_ARRAY => {
638 let v: Option<Vec<Option<time::PrimitiveDateTime>>> = row
639 .try_get(idx)
640 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
641 Ok(v.map(|items| Value::TypedArray {
642 element_type: ArrayElementType::Timestamp,
643 values: items
644 .into_iter()
645 .map(|opt| {
646 opt.map(|dt| {
647 let us = dt.assume_utc().unix_timestamp_nanos() / 1_000;
648 #[allow(clippy::cast_possible_truncation)]
649 Value::Timestamp(us as i64)
650 })
651 .unwrap_or(Value::Null)
652 })
653 .collect(),
654 })
655 .unwrap_or(Value::Null))
656 }
657 Type::DATE_ARRAY => {
659 let v: Option<Vec<Option<time::Date>>> = row
660 .try_get(idx)
661 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
662 Ok(v.map(|items| Value::TypedArray {
663 element_type: ArrayElementType::Date,
664 values: items
665 .into_iter()
666 .map(|opt| {
667 opt.map(|d| {
668 const UNIX_EPOCH_JDN: i32 = 2_440_588;
669 Value::Date(d.to_julian_day() - UNIX_EPOCH_JDN)
670 })
671 .unwrap_or(Value::Null)
672 })
673 .collect(),
674 })
675 .unwrap_or(Value::Null))
676 }
677 Type::TIME_ARRAY => {
679 let v: Option<Vec<Option<time::Time>>> = row
680 .try_get(idx)
681 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
682 Ok(v.map(|items| Value::TypedArray {
683 element_type: ArrayElementType::Time,
684 values: items
685 .into_iter()
686 .map(|opt| {
687 opt.map(|t| {
688 let (h, m, s, ns) = t.as_hms_nano();
689 let us = i64::from(h) * 3_600_000_000
690 + i64::from(m) * 60_000_000
691 + i64::from(s) * 1_000_000
692 + i64::from(ns) / 1_000;
693 Value::Time(us)
694 })
695 .unwrap_or(Value::Null)
696 })
697 .collect(),
698 })
699 .unwrap_or(Value::Null))
700 }
701 Type::UUID_ARRAY => {
703 let v: Option<Vec<Option<uuid::Uuid>>> = row
704 .try_get(idx)
705 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
706 Ok(v.map(|items| Value::TypedArray {
707 element_type: ArrayElementType::Uuid,
708 values: items
709 .into_iter()
710 .map(|opt| opt.map(|u| Value::Uuid(u.as_u128())).unwrap_or(Value::Null))
711 .collect(),
712 })
713 .unwrap_or(Value::Null))
714 }
715 Type::BYTEA_ARRAY => {
717 let v: Option<Vec<Option<Vec<u8>>>> = row
718 .try_get(idx)
719 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
720 Ok(v.map(|items| Value::TypedArray {
721 element_type: ArrayElementType::Bytea,
722 values: items
723 .into_iter()
724 .map(|opt| opt.map(Value::Blob).unwrap_or(Value::Null))
725 .collect(),
726 })
727 .unwrap_or(Value::Null))
728 }
729 Type::JSON_ARRAY => {
732 let v: Option<Vec<Option<String>>> = row
733 .try_get(idx)
734 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
735 Ok(v.map(|items| Value::TypedArray {
736 element_type: ArrayElementType::Json,
737 values: items
738 .into_iter()
739 .map(|opt| opt.map(Value::Json).unwrap_or(Value::Null))
740 .collect(),
741 })
742 .unwrap_or(Value::Null))
743 }
744 Type::JSONB_ARRAY => {
745 let v: Option<Vec<Option<String>>> = row
746 .try_get(idx)
747 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
748 Ok(v.map(|items| Value::TypedArray {
749 element_type: ArrayElementType::Jsonb,
750 values: items
751 .into_iter()
752 .map(|opt| opt.map(Value::Json).unwrap_or(Value::Null))
753 .collect(),
754 })
755 .unwrap_or(Value::Null))
756 }
757 ref other => {
759 if <String as FromSql>::accepts(other) {
760 let v: Option<String> = row
761 .try_get(idx)
762 .map_err(|e| PgError::TypeConversion(e.to_string()))?;
763 Ok(v.map(Value::Text).unwrap_or(Value::Null))
764 } else {
765 Ok(Value::Text(format!("<opaque:{}>", other.name())))
767 }
768 }
769 }
770}