faucet_source_redshift/
convert.rs

1//! Row decoding and parameter binding for the Redshift source.
2//!
3//! Redshift is PostgreSQL wire-compatible, so this mirrors the native Postgres
4//! source: values decode through `sqlx`'s Postgres row API, and JSON bind values
5//! are classified before binding so large integers keep full precision.
6
7use faucet_core::FaucetError;
8use serde_json::Value;
9use sqlx::{Column, Row};
10
11/// Convert a raw Redshift/Postgres column value to a `serde_json::Value`.
12///
13/// Tries progressively broader decodings and falls back to `Value::Null` for
14/// unsupported or SQL-NULL columns.
15pub(crate) fn pg_value_to_json(row: &sqlx::postgres::PgRow, col_name: &str) -> Value {
16    if let Ok(v) = row.try_get::<Value, _>(col_name) {
17        return v;
18    }
19    if let Ok(v) = row.try_get::<String, _>(col_name) {
20        return Value::String(v);
21    }
22    if let Ok(v) = row.try_get::<i64, _>(col_name) {
23        return Value::Number(v.into());
24    }
25    if let Ok(v) = row.try_get::<i32, _>(col_name) {
26        return Value::Number(v.into());
27    }
28    if let Ok(v) = row.try_get::<i16, _>(col_name) {
29        return Value::Number(v.into());
30    }
31    if let Ok(v) = row.try_get::<f64, _>(col_name) {
32        return serde_json::Number::from_f64(v)
33            .map(Value::Number)
34            .unwrap_or(Value::Null);
35    }
36    if let Ok(v) = row.try_get::<f32, _>(col_name) {
37        return serde_json::Number::from_f64(v as f64)
38            .map(Value::Number)
39            .unwrap_or(Value::Null);
40    }
41    if let Ok(v) = row.try_get::<bool, _>(col_name) {
42        return Value::Bool(v);
43    }
44    // Timestamps → RFC3339 / ISO-8601 strings.
45    if let Ok(v) =
46        row.try_get::<sqlx::types::chrono::DateTime<sqlx::types::chrono::Utc>, _>(col_name)
47    {
48        return Value::String(v.to_rfc3339());
49    }
50    if let Ok(v) = row.try_get::<sqlx::types::chrono::NaiveDateTime, _>(col_name) {
51        return Value::String(v.to_string());
52    }
53    if let Ok(v) = row.try_get::<sqlx::types::chrono::NaiveDate, _>(col_name) {
54        return Value::String(v.to_string());
55    }
56    if let Ok(v) = row.try_get::<sqlx::types::chrono::NaiveTime, _>(col_name) {
57        return Value::String(v.to_string());
58    }
59    if let Ok(v) = row.try_get::<sqlx::types::Uuid, _>(col_name) {
60        return Value::String(v.to_string());
61    }
62    // NUMERIC / DECIMAL → string, preserving exact precision.
63    if let Ok(v) = row.try_get::<sqlx::types::BigDecimal, _>(col_name) {
64        return Value::String(v.to_string());
65    }
66    // Binary (VARBYTE / bytea) → base64 so it survives the JSON round-trip.
67    if let Ok(v) = row.try_get::<Vec<u8>, _>(col_name) {
68        use base64::Engine as _;
69        return Value::String(base64::engine::general_purpose::STANDARD.encode(v));
70    }
71    Value::Null
72}
73
74/// Convert a single row into a JSON object keyed by column name.
75pub(crate) fn row_to_json(row: &sqlx::postgres::PgRow) -> Value {
76    let mut map = serde_json::Map::new();
77    for col in row.columns() {
78        let name = col.name().to_string();
79        let value = pg_value_to_json(row, &name);
80        map.insert(name, value);
81    }
82    Value::Object(map)
83}
84
85/// How a numeric bind value should be bound onto a `sqlx` query. Classifying
86/// before binding keeps any integer in `[i64::MIN, i64::MAX]` exact (binding
87/// large integers as `f64` silently rounds them).
88#[derive(Debug, Clone, Copy, PartialEq, Eq)]
89pub(crate) enum NumberBind {
90    /// Exact `i64`.
91    I64,
92    /// Above `i64::MAX`; bind the `u64` reinterpreted as `i64` (two's complement).
93    U64,
94    /// Genuine floating-point value.
95    F64,
96}
97
98/// Classify a JSON number into the bind category to use.
99pub(crate) fn classify_number(n: &serde_json::Number) -> NumberBind {
100    if n.is_i64() {
101        NumberBind::I64
102    } else if n.is_u64() {
103        NumberBind::U64
104    } else {
105        NumberBind::F64
106    }
107}
108
109/// Bind a slice of JSON values onto a `sqlx` query as native scalar types, in
110/// positional order (`$1, $2, …`). Binding a raw `serde_json::Value` would
111/// encode as `jsonb` and break comparisons against typed columns, so scalars
112/// are bound as their native types.
113pub(crate) fn bind_params<'q>(
114    mut query: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
115    binds: &'q [Value],
116) -> Result<sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>, FaucetError> {
117    for (i, value) in binds.iter().enumerate() {
118        query = match value {
119            Value::String(s) => query.bind(s.clone()),
120            Value::Number(n) => match classify_number(n) {
121                NumberBind::I64 => query.bind(n.as_i64().unwrap()),
122                // Above `i64::MAX`. Redshift's BIGINT is signed, and `as i64`
123                // would *wrap* — writing a large id as a large negative number, or
124                // (when this binds an incremental bookmark) comparing against a
125                // negative bound and re-reading or skipping rows. Refuse (#462).
126                NumberBind::U64 => query.bind(faucet_core::util::u64_to_signed(
127                    n.as_u64().unwrap(),
128                    &format!("bind parameter ${}", i + 1),
129                )?),
130                NumberBind::F64 => query.bind(n.as_f64().unwrap_or(0.0)),
131            },
132            Value::Bool(b) => query.bind(*b),
133            Value::Null => query.bind(None::<String>),
134            _ => query.bind(value.to_string()),
135        };
136    }
137    Ok(query)
138}
139
140#[cfg(test)]
141mod tests {
142    use super::*;
143    use serde_json::json;
144
145    fn num(v: serde_json::Value) -> serde_json::Number {
146        match v {
147            serde_json::Value::Number(n) => n,
148            _ => panic!("not a number"),
149        }
150    }
151
152    #[test]
153    fn classify_small_int_is_i64() {
154        assert_eq!(classify_number(&num(json!(42))), NumberBind::I64);
155        assert_eq!(classify_number(&num(json!(-7))), NumberBind::I64);
156    }
157
158    #[test]
159    fn classify_above_2_pow_53_stays_i64() {
160        let v = 9_007_199_254_740_993i64; // 2^53 + 1
161        assert_eq!(classify_number(&num(json!(v))), NumberBind::I64);
162    }
163
164    #[test]
165    fn classify_above_i64_max_is_u64() {
166        let v: u64 = i64::MAX as u64 + 1;
167        assert_eq!(classify_number(&num(json!(v))), NumberBind::U64);
168    }
169
170    #[test]
171    fn classify_float_is_f64() {
172        assert_eq!(classify_number(&num(json!(3.5))), NumberBind::F64);
173    }
174}
175
176#[cfg(test)]
177mod bind_overflow_tests {
178    use super::*;
179    use serde_json::json;
180
181    /// #462: Redshift's BIGINT is signed; `as i64` would wrap a large unsigned
182    /// id to a negative, and this same binder carries incremental bookmarks.
183    #[test]
184    fn u64_above_i64_max_is_refused_not_wrapped() {
185        let err = match bind_params(sqlx::query("SELECT 1"), &[json!(u64::MAX)]) {
186            Err(e) => e.to_string(),
187            Ok(_) => panic!("u64::MAX must not bind"),
188        };
189        assert!(err.contains(&u64::MAX.to_string()), "{err}");
190        assert!(
191            !err.contains("-9223372036854775808"),
192            "must not show the wrap: {err}"
193        );
194    }
195
196    #[test]
197    fn values_a_signed_column_can_hold_still_bind() {
198        for v in [json!(0), json!(-1), json!(i64::MAX), json!(i64::MAX as u64)] {
199            assert!(
200                bind_params(sqlx::query("SELECT 1"), std::slice::from_ref(&v)).is_ok(),
201                "{v} must still bind"
202            );
203        }
204    }
205}
faucet_source_redshift/convert.rs

faucet_source_redshift/
convert.rs
