diff --git a/crates/dbx-core/src/db/postgres.rs b/crates/dbx-core/src/db/postgres.rs index df26b3304..6ffd0e35f 100644 --- a/crates/dbx-core/src/db/postgres.rs +++ b/crates/dbx-core/src/db/postgres.rs @@ -151,6 +151,44 @@ impl<'a> FromSql<'a> for PgRawBytes { } } +#[derive(Clone, Copy, Debug, PartialEq)] +struct PgPoint { + x: f64, + y: f64, +} + +impl<'a> FromSql<'a> for PgPoint { + fn from_sql(_: &Type, raw: &'a [u8]) -> Result> { + decode_pg_point_bytes(raw).ok_or_else(|| "expected 16 bytes for PostgreSQL point".into()) + } + + fn accepts(ty: &Type) -> bool { + *ty == Type::POINT + } +} + +fn decode_pg_point_bytes(raw: &[u8]) -> Option { + let raw: [u8; 16] = raw.try_into().ok()?; + Some(PgPoint { + x: f64::from_be_bytes(raw[0..8].try_into().ok()?), + y: f64::from_be_bytes(raw[8..16].try_into().ok()?), + }) +} + +fn format_pg_float(value: f64) -> String { + if value == f64::INFINITY { + "Infinity".to_string() + } else if value == f64::NEG_INFINITY { + "-Infinity".to_string() + } else { + value.to_string() + } +} + +fn format_pg_point(point: PgPoint) -> String { + format!("({},{})", format_pg_float(point.x), format_pg_float(point.y)) +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] struct PgInterval { microseconds: i64, @@ -524,6 +562,7 @@ pub(crate) enum PgColType { Bytea, Json, Bool, + Point, Interval, DateRange, Temporal { fallback: PgTemporalFallback }, @@ -572,6 +611,9 @@ pub(crate) fn classify_pg_type(type_name: &str) -> PgColType { if upper == "BOOL" { return PgColType::Bool; } + if upper == "POINT" { + return PgColType::Point; + } if upper == "INTERVAL" { return PgColType::Interval; } @@ -655,6 +697,10 @@ pub(crate) fn pg_value_to_json_classified(row: &Row, idx: usize, col_type: PgCol serde_json::Value::Null } PgColType::Bool => pg_bool_value_to_json(row, idx), + PgColType::Point => row + .try_get::<_, PgPoint>(idx) + .map(|point| serde_json::Value::String(format_pg_point(point))) + .unwrap_or(serde_json::Value::Null), PgColType::Interval => row .try_get::<_, PgInterval>(idx) .map(|interval| serde_json::Value::String(format_pg_interval(interval))) @@ -5516,6 +5562,7 @@ mod tests { assert_eq!(classify_pg_type("json"), PgColType::Json); assert_eq!(classify_pg_type("JSONB"), PgColType::Json); assert_eq!(classify_pg_type("bool"), PgColType::Bool); + assert_eq!(classify_pg_type("point"), PgColType::Point); assert_eq!(classify_pg_type("timestamp"), PgColType::Temporal { fallback: PgTemporalFallback::Probe }); assert_eq!(classify_pg_type("timestamptz"), PgColType::Temporal { fallback: PgTemporalFallback::Probe }); assert_eq!(classify_pg_type("date"), PgColType::Temporal { fallback: PgTemporalFallback::Probe }); @@ -5954,6 +6001,22 @@ mod tests { assert!(!PgSystemU32::accepts(&Type::INT4)); } + #[test] + fn pg_point_decodes_binary_coordinates_as_text() { + let mut raw = Vec::new(); + raw.extend_from_slice(&(-194.0_f64).to_be_bytes()); + raw.extend_from_slice(&53.0_f64.to_be_bytes()); + + let point = PgPoint::from_sql(&Type::POINT, &raw).unwrap(); + + assert_eq!(point, PgPoint { x: -194.0, y: 53.0 }); + assert_eq!(format_pg_point(point), "(-194,53)"); + assert_eq!(format_pg_point(PgPoint { x: f64::NEG_INFINITY, y: f64::INFINITY }), "(-Infinity,Infinity)"); + assert!(PgPoint::from_sql(&Type::POINT, &[0; 15]).is_err()); + assert!(PgPoint::accepts(&Type::POINT)); + assert!(!PgPoint::accepts(&Type::BYTEA)); + } + #[test] fn pg_any_string_accepts_all_types_and_decodes_utf8() { // Accepts any type — built-in, custom enum OIDs, domains, etc. diff --git a/crates/dbx-core/tests/live_postgres_point.rs b/crates/dbx-core/tests/live_postgres_point.rs new file mode 100644 index 000000000..70426b2f3 --- /dev/null +++ b/crates/dbx-core/tests/live_postgres_point.rs @@ -0,0 +1,43 @@ +use std::time::Duration; + +use dbx_core::db::postgres; + +#[tokio::test] +#[ignore = "requires DBX_TEST_POSTGRES_URL pointing at a writable PostgreSQL database"] +async fn postgres_point_columns_match_their_text_representation() { + let url = std::env::var("DBX_TEST_POSTGRES_URL").expect("DBX_TEST_POSTGRES_URL"); + let pool = postgres::connect(&url, Duration::from_secs(5)).await.expect("connect postgres"); + let schema = format!("dbx_point_{}", std::process::id()); + let schema_ident = format!("\"{}\"", schema.replace('"', "\"\"")); + let table = format!("{schema_ident}.cities"); + + let _ = postgres::execute_query(&pool, &format!("DROP SCHEMA IF EXISTS {schema_ident} CASCADE")).await; + postgres::execute_query(&pool, &format!("CREATE SCHEMA {schema_ident}")).await.expect("create schema"); + postgres::execute_query(&pool, &format!("CREATE TABLE {table} (name text NOT NULL, location point)")) + .await + .expect("create table"); + postgres::execute_query( + &pool, + &format!("INSERT INTO {table} VALUES ('San Francisco', '(-194.0, 53.0)'), ('Unknown', NULL)"), + ) + .await + .expect("insert rows"); + + let result = postgres::execute_query( + &pool, + &format!("SELECT name, location, location::text AS location_text FROM {table} ORDER BY name"), + ) + .await + .expect("select point rows"); + + postgres::execute_query(&pool, &format!("DROP SCHEMA IF EXISTS {schema_ident} CASCADE")) + .await + .expect("drop schema"); + + assert_eq!(result.column_types, vec!["text", "point", "text"]); + assert_eq!(result.rows.len(), 2); + assert_eq!(result.rows[0][1], serde_json::json!("(-194,53)")); + assert_eq!(result.rows[0][1], result.rows[0][2]); + assert_eq!(result.rows[1][1], serde_json::Value::Null); + assert_eq!(result.rows[1][2], serde_json::Value::Null); +}