From 166027600fe89eda59c7041e0680efd4bbeb82ae Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Mon, 27 Jul 2026 15:45:04 +0800 Subject: [PATCH] fix(postgres): render interval values as text --- crates/dbx-core/src/db/postgres.rs | 117 ++++++++++++++++++++++++++++- 1 file changed, 116 insertions(+), 1 deletion(-) diff --git a/crates/dbx-core/src/db/postgres.rs b/crates/dbx-core/src/db/postgres.rs index a66829eb0..34ba6bd39 100644 --- a/crates/dbx-core/src/db/postgres.rs +++ b/crates/dbx-core/src/db/postgres.rs @@ -92,6 +92,69 @@ impl<'a> FromSql<'a> for PgRawBytes { } } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +struct PgInterval { + microseconds: i64, + days: i32, + months: i32, +} + +impl<'a> FromSql<'a> for PgInterval { + fn from_sql(_: &Type, raw: &'a [u8]) -> Result> { + decode_pg_interval_bytes(raw).ok_or_else(|| "expected 16 bytes for PostgreSQL interval".into()) + } + + fn accepts(ty: &Type) -> bool { + *ty == Type::INTERVAL + } +} + +fn decode_pg_interval_bytes(raw: &[u8]) -> Option { + let raw: [u8; 16] = raw.try_into().ok()?; + Some(PgInterval { + microseconds: i64::from_be_bytes(raw[0..8].try_into().ok()?), + days: i32::from_be_bytes(raw[8..12].try_into().ok()?), + months: i32::from_be_bytes(raw[12..16].try_into().ok()?), + }) +} + +fn push_pg_interval_component(parts: &mut Vec, value: i64, singular: &str, plural: &str) { + if value == 0 { + return; + } + let unit = if value.abs() == 1 { singular } else { plural }; + parts.push(format!("{value} {unit}")); +} + +fn format_pg_interval_time(microseconds: i64) -> String { + let signed_microseconds = i128::from(microseconds); + let sign = if signed_microseconds < 0 { "-" } else { "" }; + let absolute_microseconds = signed_microseconds.abs(); + let hours = absolute_microseconds / 3_600_000_000; + let minutes = absolute_microseconds / 60_000_000 % 60; + let seconds = absolute_microseconds / 1_000_000 % 60; + let fraction = absolute_microseconds % 1_000_000; + let mut formatted = format!("{sign}{hours:02}:{minutes:02}:{seconds:02}"); + if fraction != 0 { + let fraction = format!("{fraction:06}"); + formatted.push('.'); + formatted.push_str(fraction.trim_end_matches('0')); + } + formatted +} + +fn format_pg_interval(interval: PgInterval) -> String { + let total_months = i64::from(interval.months); + let years = total_months / 12; + let months = total_months % 12; + let mut parts = Vec::with_capacity(4); + push_pg_interval_component(&mut parts, years, "year", "years"); + push_pg_interval_component(&mut parts, months, "mon", "mons"); + push_pg_interval_component(&mut parts, i64::from(interval.days), "day", "days"); + parts.push(format_pg_interval_time(interval.microseconds)); + parts.join(" ") +} + /// Decode pgvector binary format into a Vec. /// /// pgvector binary layout (big-endian): @@ -321,6 +384,7 @@ pub(crate) enum PgColType { Bytea, Json, Bool, + Interval, Temporal { fallback: PgTemporalFallback }, Numeric, Uuid, @@ -364,6 +428,9 @@ pub(crate) fn classify_pg_type(type_name: &str) -> PgColType { if upper == "BOOL" { return PgColType::Bool; } + if upper == "INTERVAL" { + return PgColType::Interval; + } if upper.contains("TIMESTAMP") || upper == "DATE" || upper == "TIME" @@ -441,6 +508,10 @@ pub(crate) fn pg_value_to_json_classified(row: &Row, idx: usize, col_type: PgCol serde_json::Value::Null } PgColType::Bool => row.try_get::<_, bool>(idx).map(serde_json::Value::Bool).unwrap_or(serde_json::Value::Null), + PgColType::Interval => row + .try_get::<_, PgInterval>(idx) + .map(|interval| serde_json::Value::String(format_pg_interval(interval))) + .unwrap_or_else(|_| pg_fallback_value_to_json(row, idx)), PgColType::Temporal { fallback } => { if let Some(v) = pg_temporal_to_json_value(row, idx) { return v; @@ -3818,6 +3889,50 @@ mod tests { use std::time::Instant; use tokio_postgres::types::FromSql; + fn pg_interval_bytes(microseconds: i64, days: i32, months: i32) -> [u8; 16] { + let mut raw = [0_u8; 16]; + raw[0..8].copy_from_slice(µseconds.to_be_bytes()); + raw[8..12].copy_from_slice(&days.to_be_bytes()); + raw[12..16].copy_from_slice(&months.to_be_bytes()); + raw + } + + #[test] + fn postgres_interval_binary_decodes_and_formats_components() { + let microseconds = 4 * 3_600_000_000 + 5 * 60_000_000 + 6 * 1_000_000 + 123_456; + let interval = PgInterval::from_sql(&Type::INTERVAL, &pg_interval_bytes(microseconds, 3, 14)).unwrap(); + + assert_eq!(interval, PgInterval { microseconds, days: 3, months: 14 }); + assert_eq!(format_pg_interval(interval), "1 year 2 mons 3 days 04:05:06.123456"); + } + + #[test] + fn postgres_interval_formats_negative_mixed_and_zero_values() { + assert_eq!( + format_pg_interval(PgInterval { microseconds: -3_723_450_000, days: -2, months: -13 }), + "-1 year -1 mon -2 days -01:02:03.45" + ); + assert_eq!( + format_pg_interval(PgInterval { microseconds: -1, days: 2, months: -1 }), + "-1 mon 2 days -00:00:00.000001" + ); + assert_eq!(format_pg_interval(PgInterval { microseconds: 0, days: 0, months: 0 }), "00:00:00"); + } + + #[test] + fn postgres_interval_formats_now_minus_xact_start_shape() { + let elapsed = PgInterval { microseconds: 123_450_000, days: 0, months: 0 }; + assert_eq!(format_pg_interval(elapsed), "00:02:03.45"); + } + + #[test] + fn postgres_interval_rejects_invalid_binary_and_keeps_binary_protocol() { + assert!(PgInterval::from_sql(&Type::INTERVAL, &[0; 15]).is_err()); + assert_eq!(classify_pg_type("interval"), PgColType::Interval); + assert_eq!(classify_pg_type("_interval"), PgColType::Temporal { fallback: PgTemporalFallback::GenericArray }); + assert!(!pg_type_requires_text_protocol(&Type::INTERVAL, PgColType::Interval)); + } + #[test] fn postgres_custom_other_type_requires_text_protocol() { assert!(pg_scalar_type_requires_text_protocol(POSTGRES_FIRST_NORMAL_OBJECT_ID, PgColType::Other)); @@ -3912,7 +4027,7 @@ mod tests { assert_eq!(classify_pg_type("date"), PgColType::Temporal { fallback: PgTemporalFallback::Probe }); assert_eq!(classify_pg_type("time"), PgColType::Temporal { fallback: PgTemporalFallback::Probe }); assert_eq!(classify_pg_type("timetz"), PgColType::Temporal { fallback: PgTemporalFallback::Probe }); - assert_eq!(classify_pg_type("interval"), PgColType::Temporal { fallback: PgTemporalFallback::Probe }); + assert_eq!(classify_pg_type("interval"), PgColType::Interval); // 时间数组类型名在原实现中先进时间分支、解码失败后落到通用数组分支 assert_eq!(classify_pg_type("_timestamp"), PgColType::Temporal { fallback: PgTemporalFallback::GenericArray }); assert_eq!(classify_pg_type("_interval"), PgColType::Temporal { fallback: PgTemporalFallback::GenericArray });