mz_pgrepr/
value.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10use std::collections::BTreeMap;
11use std::error::Error;
12use std::{io, str};
13
14use bytes::{BufMut, BytesMut};
15use chrono::{DateTime, NaiveDateTime, NaiveTime, Utc};
16use itertools::Itertools;
17use mz_ore::cast::ReinterpretCast;
18use mz_pgwire_common::Format;
19use mz_repr::adt::array::ArrayDimension;
20use mz_repr::adt::char;
21use mz_repr::adt::date::Date;
22use mz_repr::adt::jsonb::JsonbRef;
23use mz_repr::adt::mz_acl_item::{AclItem, MzAclItem};
24use mz_repr::adt::pg_legacy_name::NAME_MAX_BYTES;
25use mz_repr::adt::range::{Range, RangeInner};
26use mz_repr::adt::timestamp::CheckedTimestamp;
27use mz_repr::strconv::{self, Nestable};
28use mz_repr::{Datum, RelationType, RowArena, RowPacker, RowRef, ScalarType};
29use postgres_types::{FromSql, IsNull, ToSql, Type as PgType};
30use uuid::Uuid;
31
32use crate::types::{UINT2, UINT4, UINT8};
33use crate::{Interval, Jsonb, Numeric, Type, UInt2, UInt4, UInt8};
34
35pub mod interval;
36pub mod jsonb;
37pub mod numeric;
38pub mod record;
39pub mod unsigned;
40
41/// A PostgreSQL datum.
42#[derive(Debug)]
43pub enum Value {
44    /// A variable-length, multi-dimensional array of values.
45    Array {
46        /// The dimensions of the array.
47        dims: Vec<ArrayDimension>,
48        /// The elements of the array.
49        elements: Vec<Option<Value>>,
50    },
51    /// A boolean value.
52    Bool(bool),
53    /// A byte array, i.e., a variable-length binary string.
54    Bytea(Vec<u8>),
55    /// A single-byte character.
56    Char(u8),
57    /// A date.
58    Date(Date),
59    /// A 4-byte floating point number.
60    Float4(f32),
61    /// An 8-byte floating point number.
62    Float8(f64),
63    /// A 2-byte signed integer.
64    Int2(i16),
65    /// A 4-byte signed integer.
66    Int4(i32),
67    /// An 8-byte signed integer.
68    Int8(i64),
69    /// A 2-byte unsigned integer.
70    UInt2(UInt2),
71    /// A 4-byte unsigned integer.
72    UInt4(UInt4),
73    /// An 8-byte unsigned integer.
74    UInt8(UInt8),
75    /// A time interval.
76    Interval(Interval),
77    /// A binary JSON blob.
78    Jsonb(Jsonb),
79    /// A sequence of homogeneous values.
80    List(Vec<Option<Value>>),
81    /// A map of string keys and homogeneous values.
82    Map(BTreeMap<String, Option<Value>>),
83    /// An identifier string of no more than 64 characters in length.
84    Name(String),
85    /// An arbitrary precision number.
86    Numeric(Numeric),
87    /// An object identifier.
88    Oid(u32),
89    /// A sequence of heterogeneous values.
90    Record(Vec<Option<Value>>),
91    /// A time.
92    Time(NaiveTime),
93    /// A date and time, without a timezone.
94    Timestamp(CheckedTimestamp<NaiveDateTime>),
95    /// A date and time, with a timezone.
96    TimestampTz(CheckedTimestamp<DateTime<Utc>>),
97    /// A variable-length string.
98    Text(String),
99    /// A fixed-length string.
100    BpChar(String),
101    /// A variable-length string with an optional limit.
102    VarChar(String),
103    /// A universally unique identifier.
104    Uuid(Uuid),
105    /// A small int vector.
106    Int2Vector {
107        /// The elements of the vector.
108        elements: Vec<Option<Value>>,
109    },
110    /// A Materialize timestamp.
111    MzTimestamp(mz_repr::Timestamp),
112    /// A contiguous range of values along a domain.
113    Range(Range<Box<Value>>),
114    /// A list of privileges granted to a role, that uses [`mz_repr::role_id::RoleId`]s for role
115    /// references.
116    MzAclItem(MzAclItem),
117    /// A list of privileges granted to a user that uses [`mz_repr::adt::system::Oid`]s for role
118    /// references. This type is used primarily for compatibility with PostgreSQL.
119    AclItem(AclItem),
120}
121
122impl Value {
123    /// Constructs a new `Value` from a Materialize datum.
124    ///
125    /// The conversion happens in the obvious manner, except that `Datum::Null`
126    /// is converted to `None` to align with how PostgreSQL handles NULL.
127    pub fn from_datum(datum: Datum, typ: &ScalarType) -> Option<Value> {
128        match (datum, typ) {
129            (Datum::Null, _) => None,
130            (Datum::True, ScalarType::Bool) => Some(Value::Bool(true)),
131            (Datum::False, ScalarType::Bool) => Some(Value::Bool(false)),
132            (Datum::Int16(i), ScalarType::Int16) => Some(Value::Int2(i)),
133            (Datum::Int32(i), ScalarType::Int32) => Some(Value::Int4(i)),
134            (Datum::Int64(i), ScalarType::Int64) => Some(Value::Int8(i)),
135            (Datum::UInt8(c), ScalarType::PgLegacyChar) => Some(Value::Char(c)),
136            (Datum::UInt16(u), ScalarType::UInt16) => Some(Value::UInt2(UInt2(u))),
137            (Datum::UInt32(oid), ScalarType::Oid) => Some(Value::Oid(oid)),
138            (Datum::UInt32(oid), ScalarType::RegClass) => Some(Value::Oid(oid)),
139            (Datum::UInt32(oid), ScalarType::RegProc) => Some(Value::Oid(oid)),
140            (Datum::UInt32(oid), ScalarType::RegType) => Some(Value::Oid(oid)),
141            (Datum::UInt32(u), ScalarType::UInt32) => Some(Value::UInt4(UInt4(u))),
142            (Datum::UInt64(u), ScalarType::UInt64) => Some(Value::UInt8(UInt8(u))),
143            (Datum::Float32(f), ScalarType::Float32) => Some(Value::Float4(*f)),
144            (Datum::Float64(f), ScalarType::Float64) => Some(Value::Float8(*f)),
145            (Datum::Numeric(d), ScalarType::Numeric { .. }) => Some(Value::Numeric(Numeric(d))),
146            (Datum::MzTimestamp(t), ScalarType::MzTimestamp) => Some(Value::MzTimestamp(t)),
147            (Datum::MzAclItem(mai), ScalarType::MzAclItem) => Some(Value::MzAclItem(mai)),
148            (Datum::AclItem(ai), ScalarType::AclItem) => Some(Value::AclItem(ai)),
149            (Datum::Date(d), ScalarType::Date) => Some(Value::Date(d)),
150            (Datum::Time(t), ScalarType::Time) => Some(Value::Time(t)),
151            (Datum::Timestamp(ts), ScalarType::Timestamp { .. }) => Some(Value::Timestamp(ts)),
152            (Datum::TimestampTz(ts), ScalarType::TimestampTz { .. }) => {
153                Some(Value::TimestampTz(ts))
154            }
155            (Datum::Interval(iv), ScalarType::Interval) => Some(Value::Interval(Interval(iv))),
156            (Datum::Bytes(b), ScalarType::Bytes) => Some(Value::Bytea(b.to_vec())),
157            (Datum::String(s), ScalarType::String) => Some(Value::Text(s.to_owned())),
158            (Datum::String(s), ScalarType::VarChar { .. }) => Some(Value::VarChar(s.to_owned())),
159            (Datum::String(s), ScalarType::Char { length }) => {
160                Some(Value::BpChar(char::format_str_pad(s, *length)))
161            }
162            (Datum::String(s), ScalarType::PgLegacyName) => Some(Value::Name(s.into())),
163            (_, ScalarType::Jsonb) => {
164                Some(Value::Jsonb(Jsonb(JsonbRef::from_datum(datum).to_owned())))
165            }
166            (Datum::Uuid(u), ScalarType::Uuid) => Some(Value::Uuid(u)),
167            (Datum::Array(array), ScalarType::Array(elem_type)) => {
168                let dims = array.dims().into_iter().collect();
169                let elements = array
170                    .elements()
171                    .iter()
172                    .map(|elem| Value::from_datum(elem, elem_type))
173                    .collect();
174                Some(Value::Array { dims, elements })
175            }
176            (Datum::Array(array), ScalarType::Int2Vector) => {
177                let dims = array.dims().into_iter();
178                assert!(dims.count() == 1, "int2vector must be 1 dimensional");
179                let elements = array
180                    .elements()
181                    .iter()
182                    .map(|elem| Value::from_datum(elem, &ScalarType::Int16))
183                    .collect();
184                Some(Value::Int2Vector { elements })
185            }
186            (Datum::List(list), ScalarType::List { element_type, .. }) => {
187                let elements = list
188                    .iter()
189                    .map(|elem| Value::from_datum(elem, element_type))
190                    .collect();
191                Some(Value::List(elements))
192            }
193            (Datum::List(record), ScalarType::Record { fields, .. }) => {
194                let fields = record
195                    .iter()
196                    .zip_eq(fields)
197                    .map(|(e, (_name, ty))| Value::from_datum(e, &ty.scalar_type))
198                    .collect();
199                Some(Value::Record(fields))
200            }
201            (Datum::Map(dict), ScalarType::Map { value_type, .. }) => {
202                let entries = dict
203                    .iter()
204                    .map(|(k, v)| (k.to_owned(), Value::from_datum(v, value_type)))
205                    .collect();
206                Some(Value::Map(entries))
207            }
208            (Datum::Range(range), ScalarType::Range { element_type }) => {
209                let value_range = range.into_bounds(|b| {
210                    Box::new(
211                        Value::from_datum(b.datum(), element_type)
212                            .expect("RangeBounds never contain Datum::Null"),
213                    )
214                });
215                Some(Value::Range(value_range))
216            }
217            _ => panic!("can't serialize {}::{:?}", datum, typ),
218        }
219    }
220
221    /// Converts a Materialize datum from this value.
222    pub fn into_datum<'a>(self, buf: &'a RowArena, typ: &Type) -> Datum<'a> {
223        match self {
224            Value::Array { dims, elements } => {
225                let element_pg_type = match typ {
226                    Type::Array(t) => &*t,
227                    _ => panic!("Value::Array should have type Type::Array. Found {:?}", typ),
228                };
229                buf.make_datum(|packer| {
230                    packer
231                        .try_push_array(
232                            &dims,
233                            elements.into_iter().map(|element| match element {
234                                Some(element) => element.into_datum(buf, element_pg_type),
235                                None => Datum::Null,
236                            }),
237                        )
238                        .unwrap();
239                })
240            }
241            Value::Int2Vector { .. } => {
242                // This situation is handled gracefully by Value::decode; if we
243                // wind up here it's a programming error.
244                unreachable!("into_datum cannot be called on Value::Int2Vector");
245            }
246            Value::Bool(true) => Datum::True,
247            Value::Bool(false) => Datum::False,
248            Value::Bytea(b) => Datum::Bytes(buf.push_bytes(b)),
249            Value::Char(c) => Datum::UInt8(c),
250            Value::Date(d) => Datum::Date(d),
251            Value::Float4(f) => Datum::Float32(f.into()),
252            Value::Float8(f) => Datum::Float64(f.into()),
253            Value::Int2(i) => Datum::Int16(i),
254            Value::Int4(i) => Datum::Int32(i),
255            Value::Int8(i) => Datum::Int64(i),
256            Value::UInt2(u) => Datum::UInt16(u.0),
257            Value::UInt4(u) => Datum::UInt32(u.0),
258            Value::UInt8(u) => Datum::UInt64(u.0),
259            Value::Jsonb(js) => buf.push_unary_row(js.0.into_row()),
260            Value::List(elems) => {
261                let elem_pg_type = match typ {
262                    Type::List(t) => &*t,
263                    _ => panic!("Value::List should have type Type::List. Found {:?}", typ),
264                };
265                buf.make_datum(|packer| {
266                    packer.push_list(elems.into_iter().map(|elem| match elem {
267                        Some(elem) => elem.into_datum(buf, elem_pg_type),
268                        None => Datum::Null,
269                    }));
270                })
271            }
272            Value::Map(map) => {
273                let elem_pg_type = match typ {
274                    Type::Map { value_type } => &*value_type,
275                    _ => panic!("Value::Map should have type Type::Map. Found {:?}", typ),
276                };
277                buf.make_datum(|packer| {
278                    packer.push_dict_with(|row| {
279                        for (k, v) in map {
280                            row.push(Datum::String(&k));
281                            row.push(match v {
282                                Some(elem) => elem.into_datum(buf, elem_pg_type),
283                                None => Datum::Null,
284                            });
285                        }
286                    });
287                })
288            }
289            Value::Oid(oid) => Datum::UInt32(oid),
290            Value::Record(_) => {
291                // This situation is handled gracefully by Value::decode; if we
292                // wind up here it's a programming error.
293                unreachable!("into_datum cannot be called on Value::Record");
294            }
295            Value::Time(t) => Datum::Time(t),
296            Value::Timestamp(ts) => Datum::Timestamp(ts),
297            Value::TimestampTz(ts) => Datum::TimestampTz(ts),
298            Value::Interval(iv) => Datum::Interval(iv.0),
299            Value::Text(s) | Value::VarChar(s) | Value::Name(s) => {
300                Datum::String(buf.push_string(s))
301            }
302            Value::BpChar(s) => Datum::String(buf.push_string(s.trim_end().into())),
303            Value::Uuid(u) => Datum::Uuid(u),
304            Value::Numeric(n) => Datum::Numeric(n.0),
305            Value::MzTimestamp(t) => Datum::MzTimestamp(t),
306            Value::Range(range) => {
307                let elem_pg_type = match typ {
308                    Type::Range { element_type } => &*element_type,
309                    _ => panic!("Value::Range should have type Type::Range. Found {:?}", typ),
310                };
311                let range = range.into_bounds(|elem| elem.into_datum(buf, elem_pg_type));
312
313                buf.make_datum(|packer| packer.push_range(range).unwrap())
314            }
315            Value::MzAclItem(mz_acl_item) => Datum::MzAclItem(mz_acl_item),
316            Value::AclItem(acl_item) => Datum::AclItem(acl_item),
317        }
318    }
319
320    /// Serializes this value to `buf` in the specified `format`.
321    pub fn encode(&self, ty: &Type, format: Format, buf: &mut BytesMut) -> Result<(), io::Error> {
322        match format {
323            Format::Text => {
324                self.encode_text(buf);
325                Ok(())
326            }
327            Format::Binary => self.encode_binary(ty, buf),
328        }
329    }
330
331    /// Serializes this value to `buf` using the [text encoding
332    /// format](Format::Text).
333    pub fn encode_text(&self, buf: &mut BytesMut) -> Nestable {
334        match self {
335            Value::Array { dims, elements } => {
336                strconv::format_array(buf, dims, elements, |buf, elem| match elem {
337                    None => Ok::<_, ()>(buf.write_null()),
338                    Some(elem) => Ok(elem.encode_text(buf.nonnull_buffer())),
339                })
340                .expect("provided closure never fails")
341            }
342            Value::Int2Vector { elements } => {
343                strconv::format_legacy_vector(buf, elements, |buf, elem| {
344                    Ok::<_, ()>(
345                        elem.as_ref()
346                            .expect("Int2Vector does not support NULL values")
347                            .encode_text(buf.nonnull_buffer()),
348                    )
349                })
350                .expect("provided closure never fails")
351            }
352            Value::Bool(b) => strconv::format_bool(buf, *b),
353            Value::Bytea(b) => strconv::format_bytes(buf, b),
354            Value::Char(c) => {
355                buf.put_u8(*c);
356                Nestable::MayNeedEscaping
357            }
358            Value::Date(d) => strconv::format_date(buf, *d),
359            Value::Int2(i) => strconv::format_int16(buf, *i),
360            Value::Int4(i) => strconv::format_int32(buf, *i),
361            Value::Int8(i) => strconv::format_int64(buf, *i),
362            Value::UInt2(u) => strconv::format_uint16(buf, u.0),
363            Value::UInt4(u) => strconv::format_uint32(buf, u.0),
364            Value::UInt8(u) => strconv::format_uint64(buf, u.0),
365            Value::Interval(iv) => strconv::format_interval(buf, iv.0),
366            Value::Float4(f) => strconv::format_float32(buf, *f),
367            Value::Float8(f) => strconv::format_float64(buf, *f),
368            Value::Jsonb(js) => strconv::format_jsonb(buf, js.0.as_ref()),
369            Value::List(elems) => strconv::format_list(buf, elems, |buf, elem| match elem {
370                None => Ok::<_, ()>(buf.write_null()),
371                Some(elem) => Ok(elem.encode_text(buf.nonnull_buffer())),
372            })
373            .expect("provided closure never fails"),
374            Value::Map(elems) => strconv::format_map(buf, elems, |buf, value| match value {
375                None => Ok::<_, ()>(buf.write_null()),
376                Some(elem) => Ok(elem.encode_text(buf.nonnull_buffer())),
377            })
378            .expect("provided closure never fails"),
379            Value::Oid(oid) => strconv::format_uint32(buf, *oid),
380            Value::Record(elems) => strconv::format_record(buf, elems, |buf, elem| match elem {
381                None => Ok::<_, ()>(buf.write_null()),
382                Some(elem) => Ok(elem.encode_text(buf.nonnull_buffer())),
383            })
384            .expect("provided closure never fails"),
385            Value::Text(s) | Value::VarChar(s) | Value::BpChar(s) | Value::Name(s) => {
386                strconv::format_string(buf, s)
387            }
388            Value::Time(t) => strconv::format_time(buf, *t),
389            Value::Timestamp(ts) => strconv::format_timestamp(buf, ts),
390            Value::TimestampTz(ts) => strconv::format_timestamptz(buf, ts),
391            Value::Uuid(u) => strconv::format_uuid(buf, *u),
392            Value::Numeric(d) => strconv::format_numeric(buf, &d.0),
393            Value::MzTimestamp(t) => strconv::format_mz_timestamp(buf, *t),
394            Value::Range(range) => strconv::format_range(buf, range, |buf, elem| match elem {
395                Some(elem) => Ok(elem.encode_text(buf.nonnull_buffer())),
396                None => Ok::<_, ()>(buf.write_null()),
397            })
398            .expect("provided closure never fails"),
399            Value::MzAclItem(mz_acl_item) => strconv::format_mz_acl_item(buf, *mz_acl_item),
400            Value::AclItem(acl_item) => strconv::format_acl_item(buf, *acl_item),
401        }
402    }
403
404    /// Serializes this value to `buf` using the [binary encoding
405    /// format](Format::Binary).
406    pub fn encode_binary(&self, ty: &Type, buf: &mut BytesMut) -> Result<(), io::Error> {
407        // NOTE: If implementing binary encoding for a previously unsupported `Value` type,
408        // please update the `can_encode_binary` method below.
409        let is_null = match self {
410            Value::Array { dims, elements } => {
411                let ndims = pg_len("number of array dimensions", dims.len())?;
412                let has_null = elements.iter().any(|e| e.is_none());
413                let elem_type = match ty {
414                    Type::Array(elem_type) => elem_type,
415                    _ => unreachable!(),
416                };
417                buf.put_i32(ndims);
418                buf.put_i32(has_null.into());
419                buf.put_u32(elem_type.oid());
420                for dim in dims {
421                    buf.put_i32(pg_len("array dimension length", dim.length)?);
422                    buf.put_i32(dim.lower_bound.try_into().map_err(|_| {
423                        io::Error::new(
424                            io::ErrorKind::Other,
425                            "array dimension lower bound does not fit into an i32",
426                        )
427                    })?);
428                }
429                for elem in elements {
430                    encode_element(buf, elem.as_ref(), elem_type)?;
431                }
432                Ok(postgres_types::IsNull::No)
433            }
434            // TODO: what is the binary format of vector types?
435            Value::Int2Vector { .. } => {
436                Err("binary encoding of int2vector is not implemented".into())
437            }
438            Value::Bool(b) => b.to_sql(&PgType::BOOL, buf),
439            Value::Bytea(b) => b.to_sql(&PgType::BYTEA, buf),
440            Value::Char(c) => i8::reinterpret_cast(*c).to_sql(&PgType::CHAR, buf),
441            Value::Date(d) => d.pg_epoch_days().to_sql(&PgType::DATE, buf),
442            Value::Float4(f) => f.to_sql(&PgType::FLOAT4, buf),
443            Value::Float8(f) => f.to_sql(&PgType::FLOAT8, buf),
444            Value::Int2(i) => i.to_sql(&PgType::INT2, buf),
445            Value::Int4(i) => i.to_sql(&PgType::INT4, buf),
446            Value::Int8(i) => i.to_sql(&PgType::INT8, buf),
447            Value::UInt2(u) => u.to_sql(&*UINT2, buf),
448            Value::UInt4(u) => u.to_sql(&*UINT4, buf),
449            Value::UInt8(u) => u.to_sql(&*UINT8, buf),
450            Value::Interval(iv) => iv.to_sql(&PgType::INTERVAL, buf),
451            Value::Jsonb(js) => js.to_sql(&PgType::JSONB, buf),
452            Value::List(_) => {
453                // A binary encoding for list is tricky. We only get one OID to
454                // describe the type of this list to the client. And we can't
455                // just up front allocate an OID for every possible list type,
456                // like PostgreSQL does for arrays, because, unlike arrays,
457                // lists can be arbitrarily nested.
458                //
459                // So, we'd need to synthesize a type with a stable OID whenever
460                // a new anonymous list type is *observed* in Materialize. Or we
461                // could mandate that only named list types can be sent over
462                // pgwire, and not anonymous list types, since named list types
463                // get a stable OID when they're created. Then we'd need to
464                // expose a table with the list OID -> element OID mapping for
465                // clients to query. And THEN we'd need to teach every client we
466                // care about how to query this table.
467                //
468                // This isn't intractible. It's how PostgreSQL's range type
469                // works, which is supported by many drivers. But our job is
470                // harder because most PostgreSQL drivers don't want to carry
471                // around code for Materialize-specific types. So we'd have to
472                // add type plugin infrastructure for those drivers, then
473                // distribute the list/map support as a plugin.
474                //
475                // Serializing the actual list would be simple, though: just a
476                // 32-bit integer describing the list length, followed by the
477                // encoding of each element in order.
478                //
479                // tl;dr it's a lot of work. For now, the recommended workaround
480                // is to either use the text encoding or convert the list to a
481                // different type (JSON, an array, unnest into rows) that does
482                // have a binary encoding.
483                Err("binary encoding of list types is not implemented".into())
484            }
485            Value::Map(_) => {
486                // Map binary encodings are hard for the same reason as list
487                // binary encodings (described above). You just have key and
488                // value OIDs to deal with rather than an element OID.
489                Err("binary encoding of map types is not implemented".into())
490            }
491            Value::Name(s) => s.to_sql(&PgType::NAME, buf),
492            Value::Oid(i) => i.to_sql(&PgType::OID, buf),
493            Value::Record(fields) => {
494                let nfields = pg_len("record field length", fields.len())?;
495                buf.put_i32(nfields);
496                let field_types = match ty {
497                    Type::Record(fields) => fields,
498                    _ => unreachable!(),
499                };
500                for (f, ty) in fields.iter().zip_eq(field_types) {
501                    buf.put_u32(ty.oid());
502                    encode_element(buf, f.as_ref(), ty)?;
503                }
504                Ok(postgres_types::IsNull::No)
505            }
506            Value::Text(s) => s.to_sql(&PgType::TEXT, buf),
507            Value::BpChar(s) => s.to_sql(&PgType::BPCHAR, buf),
508            Value::VarChar(s) => s.to_sql(&PgType::VARCHAR, buf),
509            Value::Time(t) => t.to_sql(&PgType::TIME, buf),
510            Value::Timestamp(ts) => ts.to_sql(&PgType::TIMESTAMP, buf),
511            Value::TimestampTz(ts) => ts.to_sql(&PgType::TIMESTAMPTZ, buf),
512            Value::Uuid(u) => u.to_sql(&PgType::UUID, buf),
513            Value::Numeric(a) => a.to_sql(&PgType::NUMERIC, buf),
514            Value::MzTimestamp(t) => t.to_string().to_sql(&PgType::TEXT, buf),
515            Value::Range(range) => {
516                buf.put_u8(range.pg_flag_bits());
517
518                let elem_type = match ty {
519                    Type::Range { element_type } => element_type,
520                    _ => unreachable!(),
521                };
522
523                if let Some(RangeInner { lower, upper }) = &range.inner {
524                    for bound in [&lower.bound, &upper.bound] {
525                        if let Some(bound) = bound {
526                            let base = buf.len();
527                            buf.put_i32(0);
528                            bound.encode_binary(elem_type, buf)?;
529                            let len = pg_len("encoded range bound", buf.len() - base - 4)?;
530                            buf[base..base + 4].copy_from_slice(&len.to_be_bytes());
531                        }
532                    }
533                }
534                Ok(postgres_types::IsNull::No)
535            }
536            Value::MzAclItem(mz_acl_item) => {
537                buf.extend_from_slice(&mz_acl_item.encode_binary());
538                Ok(postgres_types::IsNull::No)
539            }
540            Value::AclItem(_) => Err("aclitem has no binary encoding".into()),
541        }
542        .expect("encode_binary should never trigger a to_sql failure");
543        if let IsNull::Yes = is_null {
544            panic!("encode_binary impossibly called on a null value")
545        }
546        Ok(())
547    }
548
549    /// Static helper method to pre-validate that a given Datum corresponding to
550    /// the provided `ScalarType` can be converted into a `Value` and then encoded
551    /// as binary using `encode_binary` without an error.
552    pub fn can_encode_binary(typ: &ScalarType) -> bool {
553        match typ {
554            ScalarType::Bool => true,
555            ScalarType::Int16 => true,
556            ScalarType::Int32 => true,
557            ScalarType::Int64 => true,
558            ScalarType::PgLegacyChar => true,
559            ScalarType::UInt16 => true,
560            ScalarType::Oid => true,
561            ScalarType::RegClass => true,
562            ScalarType::RegProc => true,
563            ScalarType::RegType => true,
564            ScalarType::UInt32 => true,
565            ScalarType::UInt64 => true,
566            ScalarType::Float32 => true,
567            ScalarType::Float64 => true,
568            ScalarType::Numeric { .. } => true,
569            ScalarType::MzTimestamp => true,
570            ScalarType::MzAclItem => true,
571            ScalarType::AclItem => false, // "aclitem has no binary encoding"
572            ScalarType::Date => true,
573            ScalarType::Time => true,
574            ScalarType::Timestamp { .. } => true,
575            ScalarType::TimestampTz { .. } => true,
576            ScalarType::Interval => true,
577            ScalarType::Bytes => true,
578            ScalarType::String => true,
579            ScalarType::VarChar { .. } => true,
580            ScalarType::Char { .. } => true,
581            ScalarType::PgLegacyName => true,
582            ScalarType::Jsonb => true,
583            ScalarType::Uuid => true,
584            ScalarType::Array(elem_type) => Self::can_encode_binary(elem_type),
585            ScalarType::Int2Vector => false, // "binary encoding of int2vector is not implemented"
586            ScalarType::List { .. } => false, // "binary encoding of list types is not implemented"
587            ScalarType::Map { .. } => false, // "binary encoding of map types is not implemented"
588            ScalarType::Record { fields, .. } => fields
589                .iter()
590                .all(|(_, ty)| Self::can_encode_binary(&ty.scalar_type)),
591            ScalarType::Range { element_type } => Self::can_encode_binary(element_type),
592        }
593    }
594
595    /// Deserializes a value of type `ty` from `raw` using the specified
596    /// `format`.
597    pub fn decode(
598        format: Format,
599        ty: &Type,
600        raw: &[u8],
601    ) -> Result<Value, Box<dyn Error + Sync + Send>> {
602        match format {
603            Format::Text => Value::decode_text(ty, raw),
604            Format::Binary => Value::decode_binary(ty, raw),
605        }
606    }
607
608    /// Deserializes a value of type `ty` from `raw` using the [text encoding
609    /// format](Format::Text).
610    pub fn decode_text<'a>(
611        ty: &'a Type,
612        raw: &'a [u8],
613    ) -> Result<Value, Box<dyn Error + Sync + Send>> {
614        let s = str::from_utf8(raw)?;
615        Ok(match ty {
616            Type::Array(elem_type) => {
617                let (elements, dims) = strconv::parse_array(
618                    s,
619                    || None,
620                    |elem_text| Value::decode_text(elem_type, elem_text.as_bytes()).map(Some),
621                )?;
622                Value::Array { dims, elements }
623            }
624            Type::Int2Vector { .. } => {
625                return Err("input of Int2Vector types is not implemented".into());
626            }
627            Type::Bool => Value::Bool(strconv::parse_bool(s)?),
628            Type::Bytea => Value::Bytea(strconv::parse_bytes(s)?),
629            Type::Char => Value::Char(raw.get(0).copied().unwrap_or(0)),
630            Type::Date => Value::Date(strconv::parse_date(s)?),
631            Type::Float4 => Value::Float4(strconv::parse_float32(s)?),
632            Type::Float8 => Value::Float8(strconv::parse_float64(s)?),
633            Type::Int2 => Value::Int2(strconv::parse_int16(s)?),
634            Type::Int4 => Value::Int4(strconv::parse_int32(s)?),
635            Type::Int8 => Value::Int8(strconv::parse_int64(s)?),
636            Type::UInt2 => Value::UInt2(UInt2(strconv::parse_uint16(s)?)),
637            Type::UInt4 => Value::UInt4(UInt4(strconv::parse_uint32(s)?)),
638            Type::UInt8 => Value::UInt8(UInt8(strconv::parse_uint64(s)?)),
639            Type::Interval { .. } => Value::Interval(Interval(strconv::parse_interval(s)?)),
640            Type::Json => return Err("input of json types is not implemented".into()),
641            Type::Jsonb => Value::Jsonb(Jsonb(strconv::parse_jsonb(s)?)),
642            Type::List(elem_type) => Value::List(strconv::parse_list(
643                s,
644                matches!(**elem_type, Type::List(..)),
645                || None,
646                |elem_text| Value::decode_text(elem_type, elem_text.as_bytes()).map(Some),
647            )?),
648            Type::Map { value_type } => Value::Map(strconv::parse_map(
649                s,
650                matches!(**value_type, Type::Map { .. }),
651                |elem_text| {
652                    elem_text
653                        .map(|t| Value::decode_text(value_type, t.as_bytes()))
654                        .transpose()
655                },
656            )?),
657            Type::Name => Value::Name(strconv::parse_pg_legacy_name(s)),
658            Type::Numeric { .. } => Value::Numeric(Numeric(strconv::parse_numeric(s)?)),
659            Type::Oid | Type::RegClass | Type::RegProc | Type::RegType => {
660                Value::Oid(strconv::parse_oid(s)?)
661            }
662            Type::Record(_) => {
663                return Err("input of anonymous composite types is not implemented".into());
664            }
665            Type::Text => Value::Text(s.to_owned()),
666            Type::BpChar { .. } => Value::BpChar(s.to_owned()),
667            Type::VarChar { .. } => Value::VarChar(s.to_owned()),
668            Type::Time { .. } => Value::Time(strconv::parse_time(s)?),
669            Type::TimeTz { .. } => return Err("input of timetz types is not implemented".into()),
670            Type::Timestamp { .. } => Value::Timestamp(strconv::parse_timestamp(s)?),
671            Type::TimestampTz { .. } => Value::TimestampTz(strconv::parse_timestamptz(s)?),
672            Type::Uuid => Value::Uuid(Uuid::parse_str(s)?),
673            Type::MzTimestamp => Value::MzTimestamp(strconv::parse_mz_timestamp(s)?),
674            Type::Range { element_type } => Value::Range(strconv::parse_range(s, |elem_text| {
675                Value::decode_text(element_type, elem_text.as_bytes()).map(Box::new)
676            })?),
677            Type::MzAclItem => Value::MzAclItem(strconv::parse_mz_acl_item(s)?),
678            Type::AclItem => Value::AclItem(strconv::parse_acl_item(s)?),
679        })
680    }
681
682    /// Deserializes a value of type `ty` from `s` using the [text encoding format](Format::Text).
683    pub fn decode_text_into_row<'a>(
684        ty: &'a Type,
685        s: &'a str,
686        packer: &mut RowPacker,
687    ) -> Result<(), Box<dyn Error + Sync + Send>> {
688        Ok(match ty {
689            Type::Array(elem_type) => {
690                let (elements, dims) =
691                    strconv::parse_array(s, || None, |elem_text| Ok::<_, String>(Some(elem_text)))?;
692                // SAFETY: The function returns the number of times it called `push` on the packer.
693                unsafe {
694                    packer.push_array_with_unchecked(&dims, |packer| {
695                        let mut nelements = 0;
696                        for element in elements {
697                            match element {
698                                Some(elem_text) => {
699                                    Value::decode_text_into_row(elem_type, &elem_text, packer)?
700                                }
701
702                                None => packer.push(Datum::Null),
703                            }
704                            nelements += 1;
705                        }
706                        Ok::<_, Box<dyn Error + Sync + Send>>(nelements)
707                    })?
708                }
709            }
710            Type::Int2Vector { .. } => {
711                return Err("input of Int2Vector types is not implemented".into());
712            }
713            Type::Bool => packer.push(Datum::from(strconv::parse_bool(s)?)),
714            Type::Bytea => packer.push(Datum::Bytes(&strconv::parse_bytes(s)?)),
715            Type::Char => packer.push(Datum::UInt8(s.as_bytes().get(0).copied().unwrap_or(0))),
716            Type::Date => packer.push(Datum::Date(strconv::parse_date(s)?)),
717            Type::Float4 => packer.push(Datum::Float32(strconv::parse_float32(s)?.into())),
718            Type::Float8 => packer.push(Datum::Float64(strconv::parse_float64(s)?.into())),
719            Type::Int2 => packer.push(Datum::Int16(strconv::parse_int16(s)?)),
720            Type::Int4 => packer.push(Datum::Int32(strconv::parse_int32(s)?)),
721            Type::Int8 => packer.push(Datum::Int64(strconv::parse_int64(s)?)),
722            Type::UInt2 => packer.push(Datum::UInt16(strconv::parse_uint16(s)?)),
723            Type::UInt4 => packer.push(Datum::UInt32(strconv::parse_uint32(s)?)),
724            Type::UInt8 => packer.push(Datum::UInt64(strconv::parse_uint64(s)?)),
725            Type::Interval { .. } => packer.push(Datum::Interval(strconv::parse_interval(s)?)),
726            Type::Json => return Err("input of json types is not implemented".into()),
727            Type::Jsonb => packer.push(strconv::parse_jsonb(s)?.into_row().unpack_first()),
728            Type::List(elem_type) => {
729                let elems = strconv::parse_list(
730                    s,
731                    matches!(**elem_type, Type::List(..)),
732                    || None,
733                    |elem_text| Ok::<_, String>(Some(elem_text)),
734                )?;
735                packer.push_list_with(|packer| {
736                    for elem in elems {
737                        match elem {
738                            Some(elem) => Value::decode_text_into_row(elem_type, &elem, packer)?,
739                            None => packer.push(Datum::Null),
740                        }
741                    }
742                    Ok::<_, Box<dyn Error + Sync + Send>>(())
743                })?;
744            }
745            Type::Map { value_type } => {
746                let map =
747                    strconv::parse_map(s, matches!(**value_type, Type::Map { .. }), |elem_text| {
748                        elem_text.map(Ok::<_, String>).transpose()
749                    })?;
750                packer.push_dict_with(|row| {
751                    for (k, v) in map {
752                        row.push(Datum::String(&k));
753                        match v {
754                            Some(elem) => Value::decode_text_into_row(value_type, &elem, row)?,
755                            None => row.push(Datum::Null),
756                        }
757                    }
758                    Ok::<_, Box<dyn Error + Sync + Send>>(())
759                })?;
760            }
761            Type::Name => packer.push(Datum::String(&strconv::parse_pg_legacy_name(s))),
762            Type::Numeric { .. } => packer.push(Datum::Numeric(strconv::parse_numeric(s)?)),
763            Type::Oid | Type::RegClass | Type::RegProc | Type::RegType => {
764                packer.push(Datum::UInt32(strconv::parse_oid(s)?))
765            }
766            Type::Record(_) => {
767                return Err("input of anonymous composite types is not implemented".into());
768            }
769            Type::Text => packer.push(Datum::String(s)),
770            Type::BpChar { .. } => packer.push(Datum::String(s.trim_end())),
771            Type::VarChar { .. } => packer.push(Datum::String(s)),
772            Type::Time { .. } => packer.push(Datum::Time(strconv::parse_time(s)?)),
773            Type::TimeTz { .. } => return Err("input of timetz types is not implemented".into()),
774            Type::Timestamp { .. } => packer.push(Datum::Timestamp(strconv::parse_timestamp(s)?)),
775            Type::TimestampTz { .. } => {
776                packer.push(Datum::TimestampTz(strconv::parse_timestamptz(s)?))
777            }
778            Type::Uuid => packer.push(Datum::Uuid(Uuid::parse_str(s)?)),
779            Type::MzTimestamp => packer.push(Datum::MzTimestamp(strconv::parse_mz_timestamp(s)?)),
780            Type::Range { element_type } => {
781                let range = strconv::parse_range(s, |elem_text| {
782                    Value::decode_text(element_type, elem_text.as_bytes()).map(Box::new)
783                })?;
784                // TODO: We should be able to push ranges without scratch space, but that requires
785                // a different `push_range` API.
786                let buf = RowArena::new();
787                let range = range.into_bounds(|elem| elem.into_datum(&buf, element_type));
788
789                packer.push_range(range).unwrap()
790            }
791            Type::MzAclItem => packer.push(Datum::MzAclItem(strconv::parse_mz_acl_item(s)?)),
792            Type::AclItem => packer.push(Datum::AclItem(strconv::parse_acl_item(s)?)),
793        })
794    }
795
796    /// Deserializes a value of type `ty` from `raw` using the [binary encoding
797    /// format](Format::Binary).
798    pub fn decode_binary(ty: &Type, raw: &[u8]) -> Result<Value, Box<dyn Error + Sync + Send>> {
799        match ty {
800            Type::Array(_) => Err("input of array types is not implemented".into()),
801            Type::Int2Vector => Err("input of int2vector types is not implemented".into()),
802            Type::Bool => bool::from_sql(ty.inner(), raw).map(Value::Bool),
803            Type::Bytea => Vec::<u8>::from_sql(ty.inner(), raw).map(Value::Bytea),
804            Type::Char => {
805                i8::from_sql(ty.inner(), raw).map(|c| Value::Char(u8::reinterpret_cast(c)))
806            }
807            Type::Date => {
808                let days = i32::from_sql(ty.inner(), raw)?;
809                Ok(Value::Date(Date::from_pg_epoch(days)?))
810            }
811            Type::Float4 => f32::from_sql(ty.inner(), raw).map(Value::Float4),
812            Type::Float8 => f64::from_sql(ty.inner(), raw).map(Value::Float8),
813            Type::Int2 => i16::from_sql(ty.inner(), raw).map(Value::Int2),
814            Type::Int4 => i32::from_sql(ty.inner(), raw).map(Value::Int4),
815            Type::Int8 => i64::from_sql(ty.inner(), raw).map(Value::Int8),
816            Type::UInt2 => UInt2::from_sql(ty.inner(), raw).map(Value::UInt2),
817            Type::UInt4 => UInt4::from_sql(ty.inner(), raw).map(Value::UInt4),
818            Type::UInt8 => UInt8::from_sql(ty.inner(), raw).map(Value::UInt8),
819            Type::Interval { .. } => Interval::from_sql(ty.inner(), raw).map(Value::Interval),
820            Type::Json => Err("input of json types is not implemented".into()),
821            Type::Jsonb => Jsonb::from_sql(ty.inner(), raw).map(Value::Jsonb),
822            Type::List(_) => Err("binary decoding of list types is not implemented".into()),
823            Type::Map { .. } => Err("binary decoding of map types is not implemented".into()),
824            Type::Name => {
825                let s = String::from_sql(ty.inner(), raw)?;
826                if s.len() > NAME_MAX_BYTES {
827                    return Err("identifier too long".into());
828                }
829                Ok(Value::Name(s))
830            }
831            Type::Numeric { .. } => Numeric::from_sql(ty.inner(), raw).map(Value::Numeric),
832            Type::Oid | Type::RegClass | Type::RegProc | Type::RegType => {
833                u32::from_sql(ty.inner(), raw).map(Value::Oid)
834            }
835            Type::Record(_) => Err("input of anonymous composite types is not implemented".into()),
836            Type::Text => String::from_sql(ty.inner(), raw).map(Value::Text),
837            Type::BpChar { .. } => String::from_sql(ty.inner(), raw).map(Value::BpChar),
838            Type::VarChar { .. } => String::from_sql(ty.inner(), raw).map(Value::VarChar),
839            Type::Time { .. } => NaiveTime::from_sql(ty.inner(), raw).map(Value::Time),
840            Type::TimeTz { .. } => Err("input of timetz types is not implemented".into()),
841            Type::Timestamp { .. } => {
842                let ts = NaiveDateTime::from_sql(ty.inner(), raw)?;
843                Ok(Value::Timestamp(CheckedTimestamp::from_timestamplike(ts)?))
844            }
845            Type::TimestampTz { .. } => {
846                let ts = DateTime::<Utc>::from_sql(ty.inner(), raw)?;
847                Ok(Value::TimestampTz(CheckedTimestamp::from_timestamplike(
848                    ts,
849                )?))
850            }
851            Type::Uuid => Uuid::from_sql(ty.inner(), raw).map(Value::Uuid),
852            Type::MzTimestamp => {
853                let s = String::from_sql(ty.inner(), raw)?;
854                let t: mz_repr::Timestamp = s.parse()?;
855                Ok(Value::MzTimestamp(t))
856            }
857            Type::Range { .. } => Err("binary decoding of range types is not implemented".into()),
858            Type::MzAclItem => {
859                let mz_acl_item = MzAclItem::decode_binary(raw)?;
860                Ok(Value::MzAclItem(mz_acl_item))
861            }
862            Type::AclItem => Err("aclitem has no binary encoding".into()),
863        }
864    }
865}
866
867fn encode_element(buf: &mut BytesMut, elem: Option<&Value>, ty: &Type) -> Result<(), io::Error> {
868    match elem {
869        None => buf.put_i32(-1),
870        Some(elem) => {
871            let base = buf.len();
872            buf.put_i32(0);
873            elem.encode_binary(ty, buf)?;
874            let len = pg_len("encoded element", buf.len() - base - 4)?;
875            buf[base..base + 4].copy_from_slice(&len.to_be_bytes());
876        }
877    }
878    Ok(())
879}
880
881fn pg_len(what: &str, len: usize) -> Result<i32, io::Error> {
882    len.try_into().map_err(|_| {
883        io::Error::new(
884            io::ErrorKind::Other,
885            format!("{} does not fit into an i32", what),
886        )
887    })
888}
889
890/// Converts a Materialize row into a vector of PostgreSQL values.
891///
892/// Calling this function is equivalent to mapping [`Value::from_datum`] over
893/// every datum in `row`.
894pub fn values_from_row(row: &RowRef, typ: &RelationType) -> Vec<Option<Value>> {
895    row.iter()
896        .zip_eq(typ.column_types.iter())
897        .map(|(col, typ)| Value::from_datum(col, &typ.scalar_type))
898        .collect()
899}
900
901#[cfg(test)]
902mod tests {
903    use super::*;
904
905    /// Verifies that we correctly print the chain of parsing errors, all the way through the stack.
906    #[mz_ore::test]
907    fn decode_text_error_smoke_test() {
908        let bool_array = Value::Array {
909            dims: vec![ArrayDimension {
910                lower_bound: 0,
911                length: 1,
912            }],
913            elements: vec![Some(Value::Bool(true))],
914        };
915
916        let mut buf = BytesMut::new();
917        bool_array.encode_text(&mut buf);
918        let buf = buf.to_vec();
919
920        let int_array_tpe = Type::Array(Box::new(Type::Int4));
921        let decoded_int_array = Value::decode_text(&int_array_tpe, &buf);
922
923        assert_eq!(
924            decoded_int_array.map_err(|e| e.to_string()).unwrap_err(),
925            "invalid input syntax for type array: Specifying array lower bounds is not supported: \"[0:0]={t}\"".to_string()
926        );
927    }
928}