mz_pgrepr/

value.rs

// Copyright Materialize, Inc. and contributors. All rights reserved.
//
// Use of this software is governed by the Business Source License
// included in the LICENSE file.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0.

use std::collections::BTreeMap;
use std::error::Error;
use std::{io, str};

use bytes::{BufMut, BytesMut};
use chrono::{DateTime, NaiveDateTime, NaiveTime, Utc};
use itertools::Itertools;
use mz_ore::cast::ReinterpretCast;
use mz_pgrepr_consts::oid::TYPE_INT2_OID;
use mz_pgwire_common::Format;
use mz_repr::adt::array::ArrayDimension;
use mz_repr::adt::char;
use mz_repr::adt::date::Date;
use mz_repr::adt::jsonb::JsonbRef;
use mz_repr::adt::mz_acl_item::{AclItem, MzAclItem};
use mz_repr::adt::pg_legacy_name::NAME_MAX_BYTES;
use mz_repr::adt::range::{Range, RangeInner};
use mz_repr::adt::timestamp::CheckedTimestamp;
use mz_repr::strconv::{self, Nestable};
use mz_repr::{Datum, RowArena, RowPacker, RowRef, SqlRelationType, SqlScalarType};
use postgres_types::{FromSql, IsNull, ToSql, Type as PgType};
use uuid::Uuid;

use crate::types::{UINT2, UINT4, UINT8};
use crate::value::error::IntoDatumError;
use crate::{Interval, Jsonb, Numeric, Type, UInt2, UInt4, UInt8};

pub mod error;
pub mod interval;
pub mod jsonb;
pub mod numeric;
pub mod record;
pub mod unsigned;

/// A PostgreSQL datum.
#[derive(Debug)]
pub enum Value {
    /// A variable-length, multi-dimensional array of values.
    Array {
        /// The dimensions of the array.
        dims: Vec<ArrayDimension>,
        /// The elements of the array.
        elements: Vec<Option<Value>>,
    },
    /// A boolean value.
    Bool(bool),
    /// A byte array, i.e., a variable-length binary string.
    Bytea(Vec<u8>),
    /// A single-byte character.
    Char(u8),
    /// A date.
    Date(Date),
    /// A 4-byte floating point number.
    Float4(f32),
    /// An 8-byte floating point number.
    Float8(f64),
    /// A 2-byte signed integer.
    Int2(i16),
    /// A 4-byte signed integer.
    Int4(i32),
    /// An 8-byte signed integer.
    Int8(i64),
    /// A 2-byte unsigned integer.
    UInt2(UInt2),
    /// A 4-byte unsigned integer.
    UInt4(UInt4),
    /// An 8-byte unsigned integer.
    UInt8(UInt8),
    /// A time interval.
    Interval(Interval),
    /// A binary JSON blob.
    Jsonb(Jsonb),
    /// A sequence of homogeneous values.
    List(Vec<Option<Value>>),
    /// A map of string keys and homogeneous values.
    Map(BTreeMap<String, Option<Value>>),
    /// An identifier string of no more than 64 characters in length.
    Name(String),
    /// An arbitrary precision number.
    Numeric(Numeric),
    /// An object identifier.
    Oid(u32),
    /// A sequence of heterogeneous values.
    Record(Vec<Option<Value>>),
    /// A time.
    Time(NaiveTime),
    /// A date and time, without a timezone.
    Timestamp(CheckedTimestamp<NaiveDateTime>),
    /// A date and time, with a timezone.
    TimestampTz(CheckedTimestamp<DateTime<Utc>>),
    /// A variable-length string.
    Text(String),
    /// A fixed-length string.
    BpChar(String),
    /// A variable-length string with an optional limit.
    VarChar(String),
    /// A universally unique identifier.
    Uuid(Uuid),
    /// A small int vector.
    Int2Vector {
        /// The elements of the vector.
        elements: Vec<Option<Value>>,
    },
    /// A Materialize timestamp.
    MzTimestamp(mz_repr::Timestamp),
    /// A contiguous range of values along a domain.
    Range(Range<Box<Value>>),
    /// A list of privileges granted to a role, that uses [`mz_repr::role_id::RoleId`]s for role
    /// references.
    MzAclItem(MzAclItem),
    /// A list of privileges granted to a user that uses [`mz_repr::adt::system::Oid`]s for role
    /// references. This type is used primarily for compatibility with PostgreSQL.
    AclItem(AclItem),
}

impl Value {
    /// Constructs a new `Value` from a Materialize datum.
    ///
    /// The conversion happens in the obvious manner, except that `Datum::Null`
    /// is converted to `None` to align with how PostgreSQL handles NULL.
    pub fn from_datum(datum: Datum, typ: &SqlScalarType) -> Option<Value> {
        match (datum, typ) {
            (Datum::Null, _) => None,
            (Datum::True, SqlScalarType::Bool) => Some(Value::Bool(true)),
            (Datum::False, SqlScalarType::Bool) => Some(Value::Bool(false)),
            (Datum::Int16(i), SqlScalarType::Int16) => Some(Value::Int2(i)),
            (Datum::Int32(i), SqlScalarType::Int32) => Some(Value::Int4(i)),
            (Datum::Int64(i), SqlScalarType::Int64) => Some(Value::Int8(i)),
            (Datum::UInt8(c), SqlScalarType::PgLegacyChar) => Some(Value::Char(c)),
            (Datum::UInt16(u), SqlScalarType::UInt16) => Some(Value::UInt2(UInt2(u))),
            (Datum::UInt32(oid), SqlScalarType::Oid) => Some(Value::Oid(oid)),
            (Datum::UInt32(oid), SqlScalarType::RegClass) => Some(Value::Oid(oid)),
            (Datum::UInt32(oid), SqlScalarType::RegProc) => Some(Value::Oid(oid)),
            (Datum::UInt32(oid), SqlScalarType::RegType) => Some(Value::Oid(oid)),
            (Datum::UInt32(u), SqlScalarType::UInt32) => Some(Value::UInt4(UInt4(u))),
            (Datum::UInt64(u), SqlScalarType::UInt64) => Some(Value::UInt8(UInt8(u))),
            (Datum::Float32(f), SqlScalarType::Float32) => Some(Value::Float4(*f)),
            (Datum::Float64(f), SqlScalarType::Float64) => Some(Value::Float8(*f)),
            (Datum::Numeric(d), SqlScalarType::Numeric { .. }) => Some(Value::Numeric(Numeric(d))),
            (Datum::MzTimestamp(t), SqlScalarType::MzTimestamp) => Some(Value::MzTimestamp(t)),
            (Datum::MzAclItem(mai), SqlScalarType::MzAclItem) => Some(Value::MzAclItem(mai)),
            (Datum::AclItem(ai), SqlScalarType::AclItem) => Some(Value::AclItem(ai)),
            (Datum::Date(d), SqlScalarType::Date) => Some(Value::Date(d)),
            (Datum::Time(t), SqlScalarType::Time) => Some(Value::Time(t)),
            (Datum::Timestamp(ts), SqlScalarType::Timestamp { .. }) => Some(Value::Timestamp(ts)),
            (Datum::TimestampTz(ts), SqlScalarType::TimestampTz { .. }) => {
                Some(Value::TimestampTz(ts))
            }
            (Datum::Interval(iv), SqlScalarType::Interval) => Some(Value::Interval(Interval(iv))),
            (Datum::Bytes(b), SqlScalarType::Bytes) => Some(Value::Bytea(b.to_vec())),
            (Datum::String(s), SqlScalarType::String) => Some(Value::Text(s.to_owned())),
            (Datum::String(s), SqlScalarType::VarChar { .. }) => Some(Value::VarChar(s.to_owned())),
            (Datum::String(s), SqlScalarType::Char { length }) => {
                Some(Value::BpChar(char::format_str_pad(s, *length)))
            }
            (Datum::String(s), SqlScalarType::PgLegacyName) => Some(Value::Name(s.into())),
            (_, SqlScalarType::Jsonb) => {
                Some(Value::Jsonb(Jsonb(JsonbRef::from_datum(datum).to_owned())))
            }
            (Datum::Uuid(u), SqlScalarType::Uuid) => Some(Value::Uuid(u)),
            (Datum::Array(array), SqlScalarType::Array(elem_type)) => {
                let dims = array.dims().into_iter().collect();
                let elements = array
                    .elements()
                    .iter()
                    .map(|elem| Value::from_datum(elem, elem_type))
                    .collect();
                Some(Value::Array { dims, elements })
            }
            (Datum::Array(array), SqlScalarType::Int2Vector) => {
                assert!(
                    array.has_int2vector_dims(),
                    "int2vector must be 1 dimensional, or empty"
                );
                let elements = array
                    .elements()
                    .iter()
                    .map(|elem| Value::from_datum(elem, &SqlScalarType::Int16))
                    .collect();
                Some(Value::Int2Vector { elements })
            }
            (Datum::List(list), SqlScalarType::List { element_type, .. }) => {
                let elements = list
                    .iter()
                    .map(|elem| Value::from_datum(elem, element_type))
                    .collect();
                Some(Value::List(elements))
            }
            (Datum::List(record), SqlScalarType::Record { fields, .. }) => {
                let fields = record
                    .iter()
                    .zip_eq(fields)
                    .map(|(e, (_name, ty))| Value::from_datum(e, &ty.scalar_type))
                    .collect();
                Some(Value::Record(fields))
            }
            (Datum::Map(dict), SqlScalarType::Map { value_type, .. }) => {
                let entries = dict
                    .iter()
                    .map(|(k, v)| (k.to_owned(), Value::from_datum(v, value_type)))
                    .collect();
                Some(Value::Map(entries))
            }
            (Datum::Range(range), SqlScalarType::Range { element_type }) => {
                let value_range = range.into_bounds(|b| {
                    Box::new(
                        Value::from_datum(b.datum(), element_type)
                            .expect("RangeBounds never contain Datum::Null"),
                    )
                });
                Some(Value::Range(value_range))
            }
            _ => panic!("can't serialize {}::{:?}", datum, typ),
        }
    }

    /// Converts a Materialize datum from this value.
    pub fn into_datum<'a>(
        self,
        buf: &'a RowArena,
        typ: &Type,
    ) -> Result<Datum<'a>, IntoDatumError> {
        Ok(match self {
            Value::Array { dims, elements } => {
                let element_pg_type = match typ {
                    Type::Array(t) => &*t,
                    _ => panic!("Value::Array should have type Type::Array. Found {:?}", typ),
                };
                let elements: Result<Vec<_>, _> = elements
                    .into_iter()
                    .map(|element| match element {
                        Some(element) => element.into_datum(buf, element_pg_type),
                        None => Ok(Datum::Null),
                    })
                    .collect();
                let elements = elements?;
                buf.try_make_datum(|packer| {
                    packer
                        .try_push_array(&dims, elements)
                        .map_err(IntoDatumError::from)
                })?
            }
            Value::Int2Vector { .. } => {
                // This situation is handled gracefully by Value::decode; if we
                // wind up here it's a programming error.
                unreachable!("into_datum cannot be called on Value::Int2Vector");
            }
            Value::Bool(true) => Datum::True,
            Value::Bool(false) => Datum::False,
            Value::Bytea(b) => Datum::Bytes(buf.push_bytes(b)),
            Value::Char(c) => Datum::UInt8(c),
            Value::Date(d) => Datum::Date(d),
            Value::Float4(f) => Datum::Float32(f.into()),
            Value::Float8(f) => Datum::Float64(f.into()),
            Value::Int2(i) => Datum::Int16(i),
            Value::Int4(i) => Datum::Int32(i),
            Value::Int8(i) => Datum::Int64(i),
            Value::UInt2(u) => Datum::UInt16(u.0),
            Value::UInt4(u) => Datum::UInt32(u.0),
            Value::UInt8(u) => Datum::UInt64(u.0),
            Value::Jsonb(js) => buf.push_unary_row(js.0.into_row()),
            Value::List(elems) => {
                let elem_pg_type = match typ {
                    Type::List(t) => &*t,
                    _ => panic!("Value::List should have type Type::List. Found {:?}", typ),
                };
                let elems: Result<Vec<_>, _> = elems
                    .into_iter()
                    .map(|elem| match elem {
                        Some(elem) => elem.into_datum(buf, elem_pg_type),
                        None => Ok(Datum::Null),
                    })
                    .collect();
                let elems = elems?;
                buf.make_datum(|packer| packer.push_list(elems))
            }
            Value::Map(map) => {
                let elem_pg_type = match typ {
                    Type::Map { value_type } => &*value_type,
                    _ => panic!("Value::Map should have type Type::Map. Found {:?}", typ),
                };
                buf.try_make_datum(|packer| {
                    packer.try_push_dict_with(|row| {
                        for (k, v) in map {
                            row.push(Datum::String(buf.push_string(k)));
                            let datum = match v {
                                Some(elem) => elem.into_datum(buf, elem_pg_type)?,
                                None => Datum::Null,
                            };
                            row.push(datum);
                        }
                        Ok::<_, IntoDatumError>(())
                    })
                })?
            }
            Value::Oid(oid) => Datum::UInt32(oid),
            Value::Record(_) => {
                // This situation is handled gracefully by Value::decode; if we
                // wind up here it's a programming error.
                unreachable!("into_datum cannot be called on Value::Record");
            }
            Value::Time(t) => Datum::Time(t),
            Value::Timestamp(ts) => Datum::Timestamp(ts),
            Value::TimestampTz(ts) => Datum::TimestampTz(ts),
            Value::Interval(iv) => Datum::Interval(iv.0),
            Value::Text(s) | Value::VarChar(s) | Value::Name(s) => {
                Datum::String(buf.push_string(s))
            }
            Value::BpChar(s) => Datum::String(buf.push_string(s.trim_end().into())),
            Value::Uuid(u) => Datum::Uuid(u),
            Value::Numeric(n) => Datum::Numeric(n.0),
            Value::MzTimestamp(t) => Datum::MzTimestamp(t),
            Value::Range(range) => {
                let elem_pg_type = match typ {
                    Type::Range { element_type } => &*element_type,
                    _ => panic!("Value::Range should have type Type::Range. Found {:?}", typ),
                };
                let range = range.try_into_bounds(|elem| elem.into_datum(buf, elem_pg_type))?;
                buf.try_make_datum(|packer| packer.push_range(range).map_err(IntoDatumError::from))?
            }
            Value::MzAclItem(mz_acl_item) => Datum::MzAclItem(mz_acl_item),
            Value::AclItem(acl_item) => Datum::AclItem(acl_item),
        })
    }

    /// Like [`Self::into_datum`] but maps the error to a formatted string for decode/parameter contexts.
    ///
    /// Callers can then convert the `String` to their preferred error type (e.g. `io::Error`,
    /// protocol error). The message is `"unable to decode {context}: {error}"`.
    pub fn into_datum_decode_error<'a>(
        self,
        buf: &'a RowArena,
        typ: &Type,
        context: &str,
    ) -> Result<Datum<'a>, String> {
        self.into_datum(buf, typ)
            .map_err(|e| format!("unable to decode {}: {}", context, e))
    }

    /// Serializes this value to `buf` in the specified `format`.
    pub fn encode(&self, ty: &Type, format: Format, buf: &mut BytesMut) -> Result<(), io::Error> {
        match format {
            Format::Text => {
                self.encode_text(buf);
                Ok(())
            }
            Format::Binary => self.encode_binary(ty, buf),
        }
    }

    /// Serializes this value to `buf` using the [text encoding
    /// format](Format::Text).
    pub fn encode_text(&self, buf: &mut BytesMut) -> Nestable {
        match self {
            Value::Array { dims, elements } => {
                strconv::format_array(buf, dims, elements, |buf, elem| match elem {
                    None => Ok::<_, ()>(buf.write_null()),
                    Some(elem) => Ok(elem.encode_text(buf.nonnull_buffer())),
                })
                .expect("provided closure never fails")
            }
            Value::Int2Vector { elements } => {
                strconv::format_legacy_vector(buf, elements, |buf, elem| {
                    Ok::<_, ()>(
                        elem.as_ref()
                            .expect("Int2Vector does not support NULL values")
                            .encode_text(buf.nonnull_buffer()),
                    )
                })
                .expect("provided closure never fails")
            }
            Value::Bool(b) => strconv::format_bool(buf, *b),
            Value::Bytea(b) => strconv::format_bytes(buf, b),
            Value::Char(c) => {
                buf.put_u8(*c);
                Nestable::MayNeedEscaping
            }
            Value::Date(d) => strconv::format_date(buf, *d),
            Value::Int2(i) => strconv::format_int16(buf, *i),
            Value::Int4(i) => strconv::format_int32(buf, *i),
            Value::Int8(i) => strconv::format_int64(buf, *i),
            Value::UInt2(u) => strconv::format_uint16(buf, u.0),
            Value::UInt4(u) => strconv::format_uint32(buf, u.0),
            Value::UInt8(u) => strconv::format_uint64(buf, u.0),
            Value::Interval(iv) => strconv::format_interval(buf, iv.0),
            Value::Float4(f) => strconv::format_float32(buf, *f),
            Value::Float8(f) => strconv::format_float64(buf, *f),
            Value::Jsonb(js) => strconv::format_jsonb(buf, js.0.as_ref()),
            Value::List(elems) => strconv::format_list(buf, elems, |buf, elem| match elem {
                None => Ok::<_, ()>(buf.write_null()),
                Some(elem) => Ok(elem.encode_text(buf.nonnull_buffer())),
            })
            .expect("provided closure never fails"),
            Value::Map(elems) => strconv::format_map(buf, elems, |buf, value| match value {
                None => Ok::<_, ()>(buf.write_null()),
                Some(elem) => Ok(elem.encode_text(buf.nonnull_buffer())),
            })
            .expect("provided closure never fails"),
            Value::Oid(oid) => strconv::format_uint32(buf, *oid),
            Value::Record(elems) => strconv::format_record(buf, elems, |buf, elem| match elem {
                None => Ok::<_, ()>(buf.write_null()),
                Some(elem) => Ok(elem.encode_text(buf.nonnull_buffer())),
            })
            .expect("provided closure never fails"),
            Value::Text(s) | Value::VarChar(s) | Value::BpChar(s) | Value::Name(s) => {
                strconv::format_string(buf, s)
            }
            Value::Time(t) => strconv::format_time(buf, *t),
            Value::Timestamp(ts) => strconv::format_timestamp(buf, ts),
            Value::TimestampTz(ts) => strconv::format_timestamptz(buf, ts),
            Value::Uuid(u) => strconv::format_uuid(buf, *u),
            Value::Numeric(d) => strconv::format_numeric(buf, &d.0),
            Value::MzTimestamp(t) => strconv::format_mz_timestamp(buf, *t),
            Value::Range(range) => strconv::format_range(buf, range, |buf, elem| match elem {
                Some(elem) => Ok(elem.encode_text(buf.nonnull_buffer())),
                None => Ok::<_, ()>(buf.write_null()),
            })
            .expect("provided closure never fails"),
            Value::MzAclItem(mz_acl_item) => strconv::format_mz_acl_item(buf, *mz_acl_item),
            Value::AclItem(acl_item) => strconv::format_acl_item(buf, *acl_item),
        }
    }

    /// Serializes this value to `buf` using the [binary encoding
    /// format](Format::Binary).
    pub fn encode_binary(&self, ty: &Type, buf: &mut BytesMut) -> Result<(), io::Error> {
        // NOTE: If implementing binary encoding for a previously unsupported `Value` type,
        // please update the `binary_encoding_error` method below.
        let is_null = match self {
            Value::Array { dims, elements } => {
                let ndims = pg_len("number of array dimensions", dims.len())?;
                let has_null = elements.iter().any(|e| e.is_none());
                let elem_type = match ty {
                    Type::Array(elem_type) => elem_type,
                    _ => unreachable!(),
                };
                buf.put_i32(ndims);
                buf.put_i32(has_null.into());
                buf.put_u32(elem_type.oid());
                for dim in dims {
                    buf.put_i32(pg_len("array dimension length", dim.length)?);
                    buf.put_i32(dim.lower_bound.try_into().map_err(|_| {
                        io::Error::new(
                            io::ErrorKind::InvalidData,
                            "array dimension lower bound does not fit into an i32",
                        )
                    })?);
                }
                for elem in elements {
                    encode_element(buf, elem.as_ref(), elem_type)?;
                }
                Ok(postgres_types::IsNull::No)
            }
            Value::Int2Vector { elements } => {
                // this should always be `false`, but there are exceptions in postgres
                // feels better to compute this than to assert otherwise
                let has_null = elements.iter().any(|e| e.is_none());
                buf.put_i32(1);
                buf.put_i32(has_null.into());
                buf.put_u32(TYPE_INT2_OID);
                buf.put_i32(pg_len("int2vector dimension length", elements.len())?);
                buf.put_i32(0);
                for elem in elements {
                    encode_element(buf, elem.as_ref(), &Type::Int2)?;
                }
                Ok(postgres_types::IsNull::No)
            }
            Value::Bool(b) => b.to_sql(&PgType::BOOL, buf),
            Value::Bytea(b) => b.to_sql(&PgType::BYTEA, buf),
            Value::Char(c) => i8::reinterpret_cast(*c).to_sql(&PgType::CHAR, buf),
            Value::Date(d) => d.pg_epoch_days().to_sql(&PgType::DATE, buf),
            Value::Float4(f) => f.to_sql(&PgType::FLOAT4, buf),
            Value::Float8(f) => f.to_sql(&PgType::FLOAT8, buf),
            Value::Int2(i) => i.to_sql(&PgType::INT2, buf),
            Value::Int4(i) => i.to_sql(&PgType::INT4, buf),
            Value::Int8(i) => i.to_sql(&PgType::INT8, buf),
            Value::UInt2(u) => u.to_sql(&*UINT2, buf),
            Value::UInt4(u) => u.to_sql(&*UINT4, buf),
            Value::UInt8(u) => u.to_sql(&*UINT8, buf),
            Value::Interval(iv) => iv.to_sql(&PgType::INTERVAL, buf),
            Value::Jsonb(js) => js.to_sql(&PgType::JSONB, buf),
            Value::List(_) => {
                // A binary encoding for list is tricky. We only get one OID to
                // describe the type of this list to the client. And we can't
                // just up front allocate an OID for every possible list type,
                // like PostgreSQL does for arrays, because, unlike arrays,
                // lists can be arbitrarily nested.
                //
                // So, we'd need to synthesize a type with a stable OID whenever
                // a new anonymous list type is *observed* in Materialize. Or we
                // could mandate that only named list types can be sent over
                // pgwire, and not anonymous list types, since named list types
                // get a stable OID when they're created. Then we'd need to
                // expose a table with the list OID -> element OID mapping for
                // clients to query. And THEN we'd need to teach every client we
                // care about how to query this table.
                //
                // This isn't intractible. It's how PostgreSQL's range type
                // works, which is supported by many drivers. But our job is
                // harder because most PostgreSQL drivers don't want to carry
                // around code for Materialize-specific types. So we'd have to
                // add type plugin infrastructure for those drivers, then
                // distribute the list/map support as a plugin.
                //
                // Serializing the actual list would be simple, though: just a
                // 32-bit integer describing the list length, followed by the
                // encoding of each element in order.
                //
                // tl;dr it's a lot of work. For now, the recommended workaround
                // is to either use the text encoding or convert the list to a
                // different type (JSON, an array, unnest into rows) that does
                // have a binary encoding.
                Err("binary encoding of list types is not implemented".into())
            }
            Value::Map(_) => {
                // Map binary encodings are hard for the same reason as list
                // binary encodings (described above). You just have key and
                // value OIDs to deal with rather than an element OID.
                Err("binary encoding of map types is not implemented".into())
            }
            Value::Name(s) => s.to_sql(&PgType::NAME, buf),
            Value::Oid(i) => i.to_sql(&PgType::OID, buf),
            Value::Record(fields) => {
                let nfields = pg_len("record field length", fields.len())?;
                buf.put_i32(nfields);
                let field_types = match ty {
                    Type::Record(fields) => fields,
                    _ => unreachable!(),
                };
                for (f, ty) in fields.iter().zip_eq(field_types) {
                    buf.put_u32(ty.oid());
                    encode_element(buf, f.as_ref(), ty)?;
                }
                Ok(postgres_types::IsNull::No)
            }
            Value::Text(s) => s.to_sql(&PgType::TEXT, buf),
            Value::BpChar(s) => s.to_sql(&PgType::BPCHAR, buf),
            Value::VarChar(s) => s.to_sql(&PgType::VARCHAR, buf),
            Value::Time(t) => t.to_sql(&PgType::TIME, buf),
            Value::Timestamp(ts) => ts.to_sql(&PgType::TIMESTAMP, buf),
            Value::TimestampTz(ts) => ts.to_sql(&PgType::TIMESTAMPTZ, buf),
            Value::Uuid(u) => u.to_sql(&PgType::UUID, buf),
            Value::Numeric(a) => a.to_sql(&PgType::NUMERIC, buf),
            Value::MzTimestamp(t) => t.to_string().to_sql(&PgType::TEXT, buf),
            Value::Range(range) => {
                buf.put_u8(range.pg_flag_bits());

                let elem_type = match ty {
                    Type::Range { element_type } => element_type,
                    _ => unreachable!(),
                };

                if let Some(RangeInner { lower, upper }) = &range.inner {
                    for bound in [&lower.bound, &upper.bound] {
                        if let Some(bound) = bound {
                            let base = buf.len();
                            buf.put_i32(0);
                            bound.encode_binary(elem_type, buf)?;
                            let len = pg_len("encoded range bound", buf.len() - base - 4)?;
                            buf[base..base + 4].copy_from_slice(&len.to_be_bytes());
                        }
                    }
                }
                Ok(postgres_types::IsNull::No)
            }
            Value::MzAclItem(mz_acl_item) => {
                buf.extend_from_slice(&mz_acl_item.encode_binary());
                Ok(postgres_types::IsNull::No)
            }
            Value::AclItem(_) => Err("aclitem has no binary encoding".into()),
        }
        .expect("encode_binary should never trigger a to_sql failure");
        if let IsNull::Yes = is_null {
            panic!("encode_binary impossibly called on a null value")
        }
        Ok(())
    }

    /// Static helper method to pre-validate that a Datum corresponding to
    /// the provided `SqlScalarType` can be converted into a `Value` and then
    /// encoded as binary using `encode_binary` without an error.
    ///
    /// Returns `Ok(())` if the type (including all of its nested element/field
    /// types) supports binary encoding, or `Err(reason)` describing the first
    /// unsupported type encountered. Container types are checked recursively, so
    /// e.g. a record or array that contains a `list` is rejected.
    ///
    /// The error messages mirror PostgreSQL's `no binary output function
    /// available for type <t>` so that drivers and users see a familiar
    /// diagnostic. Callers should report these errors with the SQLSTATE
    /// PostgreSQL uses for the same condition, `42883` (undefined_function).
    pub fn binary_encoding_error(typ: &SqlScalarType) -> Result<(), &'static str> {
        match typ {
            SqlScalarType::Bool => Ok(()),
            SqlScalarType::Int16 => Ok(()),
            SqlScalarType::Int32 => Ok(()),
            SqlScalarType::Int64 => Ok(()),
            SqlScalarType::PgLegacyChar => Ok(()),
            SqlScalarType::UInt16 => Ok(()),
            SqlScalarType::Oid => Ok(()),
            SqlScalarType::RegClass => Ok(()),
            SqlScalarType::RegProc => Ok(()),
            SqlScalarType::RegType => Ok(()),
            SqlScalarType::UInt32 => Ok(()),
            SqlScalarType::UInt64 => Ok(()),
            SqlScalarType::Float32 => Ok(()),
            SqlScalarType::Float64 => Ok(()),
            SqlScalarType::Numeric { .. } => Ok(()),
            SqlScalarType::MzTimestamp => Ok(()),
            SqlScalarType::MzAclItem => Ok(()),
            SqlScalarType::AclItem => Err("no binary output function available for type aclitem"),
            SqlScalarType::Date => Ok(()),
            SqlScalarType::Time => Ok(()),
            SqlScalarType::Timestamp { .. } => Ok(()),
            SqlScalarType::TimestampTz { .. } => Ok(()),
            SqlScalarType::Interval => Ok(()),
            SqlScalarType::Bytes => Ok(()),
            SqlScalarType::String => Ok(()),
            SqlScalarType::VarChar { .. } => Ok(()),
            SqlScalarType::Char { .. } => Ok(()),
            SqlScalarType::PgLegacyName => Ok(()),
            SqlScalarType::Jsonb => Ok(()),
            SqlScalarType::Uuid => Ok(()),
            SqlScalarType::Array(elem_type) => Self::binary_encoding_error(elem_type),
            SqlScalarType::Int2Vector => Ok(()),
            SqlScalarType::List { .. } => Err("no binary output function available for type list"),
            SqlScalarType::Map { .. } => Err("no binary output function available for type map"),
            SqlScalarType::Record { fields, .. } => fields
                .iter()
                .try_for_each(|(_, ty)| Self::binary_encoding_error(&ty.scalar_type)),
            SqlScalarType::Range { element_type } => Self::binary_encoding_error(element_type),
        }
    }

    /// Returns whether a value of the given `SqlScalarType` can be encoded using
    /// the binary format. See [`Value::binary_encoding_error`] for details,
    /// including the (recursive) handling of container types.
    pub fn can_encode_binary(typ: &SqlScalarType) -> bool {
        Self::binary_encoding_error(typ).is_ok()
    }

    /// Deserializes a value of type `ty` from `raw` using the specified
    /// `format`.
    pub fn decode(
        format: Format,
        ty: &Type,
        raw: &[u8],
    ) -> Result<Value, Box<dyn Error + Sync + Send>> {
        match format {
            Format::Text => Value::decode_text(ty, raw),
            Format::Binary => Value::decode_binary(ty, raw),
        }
    }

    /// Deserializes a value of type `ty` from `raw` using the [text encoding
    /// format](Format::Text).
    pub fn decode_text<'a>(
        ty: &'a Type,
        raw: &'a [u8],
    ) -> Result<Value, Box<dyn Error + Sync + Send>> {
        let s = str::from_utf8(raw)?;
        Ok(match ty {
            Type::Array(elem_type) => {
                let (elements, dims) = strconv::parse_array(
                    s,
                    || None,
                    |elem_text| Value::decode_text(elem_type, elem_text.as_bytes()).map(Some),
                )?;
                Value::Array { dims, elements }
            }
            Type::Int2Vector { .. } => {
                return Err("input of Int2Vector types is not implemented".into());
            }
            Type::Bool => Value::Bool(strconv::parse_bool(s)?),
            Type::Bytea => Value::Bytea(strconv::parse_bytes(s)?),
            Type::Char => Value::Char(raw.get(0).copied().unwrap_or(0)),
            Type::Date => Value::Date(strconv::parse_date(s)?),
            Type::Float4 => Value::Float4(strconv::parse_float32(s)?),
            Type::Float8 => Value::Float8(strconv::parse_float64(s)?),
            Type::Int2 => Value::Int2(strconv::parse_int16(s)?),
            Type::Int4 => Value::Int4(strconv::parse_int32(s)?),
            Type::Int8 => Value::Int8(strconv::parse_int64(s)?),
            Type::UInt2 => Value::UInt2(UInt2(strconv::parse_uint16(s)?)),
            Type::UInt4 => Value::UInt4(UInt4(strconv::parse_uint32(s)?)),
            Type::UInt8 => Value::UInt8(UInt8(strconv::parse_uint64(s)?)),
            Type::Interval { .. } => Value::Interval(Interval(strconv::parse_interval(s)?)),
            Type::Json => return Err("input of json types is not implemented".into()),
            Type::Jsonb => Value::Jsonb(Jsonb(strconv::parse_jsonb(s)?)),
            Type::List(elem_type) => Value::List(strconv::parse_list(
                s,
                matches!(**elem_type, Type::List(..)),
                || None,
                |elem_text| Value::decode_text(elem_type, elem_text.as_bytes()).map(Some),
            )?),
            Type::Map { value_type } => Value::Map(strconv::parse_map(
                s,
                matches!(**value_type, Type::Map { .. }),
                |elem_text| {
                    elem_text
                        .map(|t| Value::decode_text(value_type, t.as_bytes()))
                        .transpose()
                },
            )?),
            Type::Name => Value::Name(strconv::parse_pg_legacy_name(s)),
            Type::Numeric { .. } => Value::Numeric(Numeric(strconv::parse_numeric(s)?)),
            Type::Oid | Type::RegClass | Type::RegProc | Type::RegType => {
                Value::Oid(strconv::parse_oid(s)?)
            }
            Type::Record(_) => {
                return Err("input of anonymous composite types is not implemented".into());
            }
            Type::Text => Value::Text(s.to_owned()),
            Type::BpChar { .. } => Value::BpChar(s.to_owned()),
            Type::VarChar { .. } => Value::VarChar(s.to_owned()),
            Type::Time { .. } => Value::Time(strconv::parse_time(s)?),
            Type::TimeTz { .. } => return Err("input of timetz types is not implemented".into()),
            Type::Timestamp { .. } => Value::Timestamp(strconv::parse_timestamp(s)?),
            Type::TimestampTz { .. } => Value::TimestampTz(strconv::parse_timestamptz(s)?),
            Type::Uuid => Value::Uuid(Uuid::parse_str(s)?),
            Type::MzTimestamp => Value::MzTimestamp(strconv::parse_mz_timestamp(s)?),
            Type::Range { element_type } => Value::Range(strconv::parse_range(s, |elem_text| {
                Value::decode_text(element_type, elem_text.as_bytes()).map(Box::new)
            })?),
            Type::MzAclItem => Value::MzAclItem(strconv::parse_mz_acl_item(s)?),
            Type::AclItem => Value::AclItem(strconv::parse_acl_item(s)?),
        })
    }

    /// Deserializes a value of type `ty` from `s` using the [text encoding format](Format::Text).
    pub fn decode_text_into_row<'a>(
        ty: &'a Type,
        s: &'a str,
        packer: &mut RowPacker,
    ) -> Result<(), Box<dyn Error + Sync + Send>> {
        Ok(match ty {
            Type::Array(elem_type) => {
                let (elements, dims) =
                    strconv::parse_array(s, || None, |elem_text| Ok::<_, String>(Some(elem_text)))?;
                // SAFETY: The function returns the number of times it called `push` on the packer.
                unsafe {
                    packer.push_array_with_unchecked(&dims, |packer| {
                        let mut nelements = 0;
                        for element in elements {
                            match element {
                                Some(elem_text) => {
                                    Value::decode_text_into_row(elem_type, &elem_text, packer)?
                                }

                                None => packer.push(Datum::Null),
                            }
                            nelements += 1;
                        }
                        Ok::<_, Box<dyn Error + Sync + Send>>(nelements)
                    })?
                }
            }
            Type::Int2Vector { .. } => {
                return Err("input of Int2Vector types is not implemented".into());
            }
            Type::Bool => packer.push(Datum::from(strconv::parse_bool(s)?)),
            Type::Bytea => packer.push(Datum::Bytes(&strconv::parse_bytes(s)?)),
            Type::Char => packer.push(Datum::UInt8(s.as_bytes().get(0).copied().unwrap_or(0))),
            Type::Date => packer.push(Datum::Date(strconv::parse_date(s)?)),
            Type::Float4 => packer.push(Datum::Float32(strconv::parse_float32(s)?.into())),
            Type::Float8 => packer.push(Datum::Float64(strconv::parse_float64(s)?.into())),
            Type::Int2 => packer.push(Datum::Int16(strconv::parse_int16(s)?)),
            Type::Int4 => packer.push(Datum::Int32(strconv::parse_int32(s)?)),
            Type::Int8 => packer.push(Datum::Int64(strconv::parse_int64(s)?)),
            Type::UInt2 => packer.push(Datum::UInt16(strconv::parse_uint16(s)?)),
            Type::UInt4 => packer.push(Datum::UInt32(strconv::parse_uint32(s)?)),
            Type::UInt8 => packer.push(Datum::UInt64(strconv::parse_uint64(s)?)),
            Type::Interval { .. } => packer.push(Datum::Interval(strconv::parse_interval(s)?)),
            Type::Json => return Err("input of json types is not implemented".into()),
            Type::Jsonb => packer.push(strconv::parse_jsonb(s)?.into_row().unpack_first()),
            Type::List(elem_type) => {
                let elems = strconv::parse_list(
                    s,
                    matches!(**elem_type, Type::List(..)),
                    || None,
                    |elem_text| Ok::<_, String>(Some(elem_text)),
                )?;
                packer.push_list_with(|packer| {
                    for elem in elems {
                        match elem {
                            Some(elem) => Value::decode_text_into_row(elem_type, &elem, packer)?,
                            None => packer.push(Datum::Null),
                        }
                    }
                    Ok::<_, Box<dyn Error + Sync + Send>>(())
                })?;
            }
            Type::Map { value_type } => {
                let map =
                    strconv::parse_map(s, matches!(**value_type, Type::Map { .. }), |elem_text| {
                        elem_text.map(Ok::<_, String>).transpose()
                    })?;
                packer.push_dict_with(|row| {
                    for (k, v) in map {
                        row.push(Datum::String(&k));
                        match v {
                            Some(elem) => Value::decode_text_into_row(value_type, &elem, row)?,
                            None => row.push(Datum::Null),
                        }
                    }
                    Ok::<_, Box<dyn Error + Sync + Send>>(())
                })?;
            }
            Type::Name => packer.push(Datum::String(&strconv::parse_pg_legacy_name(s))),
            Type::Numeric { .. } => packer.push(Datum::Numeric(strconv::parse_numeric(s)?)),
            Type::Oid | Type::RegClass | Type::RegProc | Type::RegType => {
                packer.push(Datum::UInt32(strconv::parse_oid(s)?))
            }
            Type::Record(_) => {
                return Err("input of anonymous composite types is not implemented".into());
            }
            Type::Text => packer.push(Datum::String(s)),
            Type::BpChar { .. } => packer.push(Datum::String(s.trim_end())),
            Type::VarChar { .. } => packer.push(Datum::String(s)),
            Type::Time { .. } => packer.push(Datum::Time(strconv::parse_time(s)?)),
            Type::TimeTz { .. } => return Err("input of timetz types is not implemented".into()),
            Type::Timestamp { .. } => packer.push(Datum::Timestamp(strconv::parse_timestamp(s)?)),
            Type::TimestampTz { .. } => {
                packer.push(Datum::TimestampTz(strconv::parse_timestamptz(s)?))
            }
            Type::Uuid => packer.push(Datum::Uuid(Uuid::parse_str(s)?)),
            Type::MzTimestamp => packer.push(Datum::MzTimestamp(strconv::parse_mz_timestamp(s)?)),
            Type::Range { element_type } => {
                let range = strconv::parse_range(s, |elem_text| {
                    Value::decode_text(element_type, elem_text.as_bytes()).map(Box::new)
                })?;
                // TODO: We should be able to push ranges without scratch space, but that requires
                // a different `push_range` API.
                let buf = RowArena::new();
                let range = range
                    .try_into_bounds(|elem| elem.into_datum(&buf, element_type))
                    .map_err(Box::<dyn Error + Sync + Send>::from)?;
                packer
                    .push_range(range)
                    .map_err(Box::<dyn Error + Sync + Send>::from)?;
            }
            Type::MzAclItem => packer.push(Datum::MzAclItem(strconv::parse_mz_acl_item(s)?)),
            Type::AclItem => packer.push(Datum::AclItem(strconv::parse_acl_item(s)?)),
        })
    }

    /// Deserializes a value of type `ty` from `raw` using the [binary encoding
    /// format](Format::Binary).
    pub fn decode_binary(ty: &Type, raw: &[u8]) -> Result<Value, Box<dyn Error + Sync + Send>> {
        match ty {
            Type::Array(_) => Err("input of array types is not implemented".into()),
            Type::Int2Vector => Err("input of int2vector types is not implemented".into()),
            Type::Bool => bool::from_sql(ty.inner(), raw).map(Value::Bool),
            Type::Bytea => Vec::<u8>::from_sql(ty.inner(), raw).map(Value::Bytea),
            Type::Char => {
                i8::from_sql(ty.inner(), raw).map(|c| Value::Char(u8::reinterpret_cast(c)))
            }
            Type::Date => {
                let days = i32::from_sql(ty.inner(), raw)?;
                Ok(Value::Date(Date::from_pg_epoch(days)?))
            }
            Type::Float4 => f32::from_sql(ty.inner(), raw).map(Value::Float4),
            Type::Float8 => f64::from_sql(ty.inner(), raw).map(Value::Float8),
            Type::Int2 => i16::from_sql(ty.inner(), raw).map(Value::Int2),
            Type::Int4 => i32::from_sql(ty.inner(), raw).map(Value::Int4),
            Type::Int8 => i64::from_sql(ty.inner(), raw).map(Value::Int8),
            Type::UInt2 => UInt2::from_sql(ty.inner(), raw).map(Value::UInt2),
            Type::UInt4 => UInt4::from_sql(ty.inner(), raw).map(Value::UInt4),
            Type::UInt8 => UInt8::from_sql(ty.inner(), raw).map(Value::UInt8),
            Type::Interval { .. } => Interval::from_sql(ty.inner(), raw).map(Value::Interval),
            Type::Json => Err("input of json types is not implemented".into()),
            Type::Jsonb => Jsonb::from_sql(ty.inner(), raw).map(Value::Jsonb),
            Type::List(_) => Err("binary decoding of list types is not implemented".into()),
            Type::Map { .. } => Err("binary decoding of map types is not implemented".into()),
            Type::Name => {
                let s = String::from_sql(ty.inner(), raw)?;
                if s.len() > NAME_MAX_BYTES {
                    return Err("identifier too long".into());
                }
                Ok(Value::Name(s))
            }
            Type::Numeric { .. } => Numeric::from_sql(ty.inner(), raw).map(Value::Numeric),
            Type::Oid | Type::RegClass | Type::RegProc | Type::RegType => {
                u32::from_sql(ty.inner(), raw).map(Value::Oid)
            }
            Type::Record(_) => Err("input of anonymous composite types is not implemented".into()),
            Type::Text => String::from_sql(ty.inner(), raw).map(Value::Text),
            Type::BpChar { .. } => String::from_sql(ty.inner(), raw).map(Value::BpChar),
            Type::VarChar { .. } => String::from_sql(ty.inner(), raw).map(Value::VarChar),
            Type::Time { .. } => NaiveTime::from_sql(ty.inner(), raw).map(Value::Time),
            Type::TimeTz { .. } => Err("input of timetz types is not implemented".into()),
            Type::Timestamp { .. } => {
                let ts = NaiveDateTime::from_sql(ty.inner(), raw)?;
                Ok(Value::Timestamp(CheckedTimestamp::from_timestamplike(ts)?))
            }
            Type::TimestampTz { .. } => {
                let ts = DateTime::<Utc>::from_sql(ty.inner(), raw)?;
                Ok(Value::TimestampTz(CheckedTimestamp::from_timestamplike(
                    ts,
                )?))
            }
            Type::Uuid => Uuid::from_sql(ty.inner(), raw).map(Value::Uuid),
            Type::MzTimestamp => {
                let s = String::from_sql(ty.inner(), raw)?;
                let t: mz_repr::Timestamp = s.parse()?;
                Ok(Value::MzTimestamp(t))
            }
            Type::Range { .. } => Err("binary decoding of range types is not implemented".into()),
            Type::MzAclItem => {
                let mz_acl_item = MzAclItem::decode_binary(raw)?;
                Ok(Value::MzAclItem(mz_acl_item))
            }
            Type::AclItem => Err("aclitem has no binary encoding".into()),
        }
    }
}

fn encode_element(buf: &mut BytesMut, elem: Option<&Value>, ty: &Type) -> Result<(), io::Error> {
    match elem {
        None => buf.put_i32(-1),
        Some(elem) => {
            let base = buf.len();
            buf.put_i32(0);
            elem.encode_binary(ty, buf)?;
            let len = pg_len("encoded element", buf.len() - base - 4)?;
            buf[base..base + 4].copy_from_slice(&len.to_be_bytes());
        }
    }
    Ok(())
}

fn pg_len(what: &str, len: usize) -> Result<i32, io::Error> {
    len.try_into().map_err(|_| {
        io::Error::new(
            io::ErrorKind::InvalidData,
            format!("{} does not fit into an i32", what),
        )
    })
}

/// Converts a Materialize row into a vector of PostgreSQL values.
///
/// Calling this function is equivalent to mapping [`Value::from_datum`] over
/// every datum in `row`.
pub fn values_from_row(row: &RowRef, typ: &SqlRelationType) -> Vec<Option<Value>> {
    row.iter()
        .zip_eq(typ.column_types.iter())
        .map(|(col, typ)| Value::from_datum(col, &typ.scalar_type))
        .collect()
}

#[cfg(test)]
mod tests {
    use mz_repr::arb_datum_for_scalar;
    use proptest::prelude::*;

    use super::*;

    /// Property test: [`Value::binary_encoding_error`] agrees with the actual
    /// behavior of [`Value::encode_binary`] for every `(SqlScalarType, Datum)`
    /// pair the proptest infrastructure can generate.
    ///
    /// This guards against future drift between the static predicate and the
    /// encoder: any new `SqlScalarType` variant whose classification disagrees
    /// with what `encode_binary` actually does will surface here.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // numeric/decimal contexts unsupported under miri
    fn proptest_binary_encoding_error_matches_encode_binary() {
        let strat =
            any::<SqlScalarType>().prop_flat_map(|ty| (Just(ty.clone()), arb_datum_for_scalar(ty)));
        proptest!(ProptestConfig::with_cases(256), |((ty, prop_datum) in strat)| {
            // `binary_encoding_error` is a precondition for callers of
            // `encode_binary`: if it returns `Ok`, then encoding must succeed
            // (and not panic via the internal `.expect`).
            if Value::binary_encoding_error(&ty).is_err() {
                return Ok(());
            }
            let datum = Datum::from(&prop_datum);
            let value = match Value::from_datum(datum, &ty) {
                Some(v) => v,
                // `Datum::Null` produces `None`; nothing to encode.
                None => return Ok(()),
            };
            let pg_ty = Type::from(&ty);
            let mut buf = BytesMut::new();
            value
                .encode_binary(&pg_ty, &mut buf)
                .expect("encode_binary must succeed when binary_encoding_error returns Ok");
        });
    }

    /// Verifies that we correctly print the chain of parsing errors, all the way through the stack.
    #[mz_ore::test]
    fn decode_text_error_smoke_test() {
        let bool_array = Value::Array {
            dims: vec![ArrayDimension {
                lower_bound: 0,
                length: 1,
            }],
            elements: vec![Some(Value::Bool(true))],
        };

        let mut buf = BytesMut::new();
        bool_array.encode_text(&mut buf);
        let buf = buf.to_vec();

        let int_array_tpe = Type::Array(Box::new(Type::Int4));
        let decoded_int_array = Value::decode_text(&int_array_tpe, &buf);

        assert_eq!(
            decoded_int_array.map_err(|e| e.to_string()).unwrap_err(),
            "invalid input syntax for type array: Specifying array lower bounds is not supported: \"[0:0]={t}\"".to_string()
        );
    }
}