1use 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#[derive(Debug)]
43pub enum Value {
44 Array {
46 dims: Vec<ArrayDimension>,
48 elements: Vec<Option<Value>>,
50 },
51 Bool(bool),
53 Bytea(Vec<u8>),
55 Char(u8),
57 Date(Date),
59 Float4(f32),
61 Float8(f64),
63 Int2(i16),
65 Int4(i32),
67 Int8(i64),
69 UInt2(UInt2),
71 UInt4(UInt4),
73 UInt8(UInt8),
75 Interval(Interval),
77 Jsonb(Jsonb),
79 List(Vec<Option<Value>>),
81 Map(BTreeMap<String, Option<Value>>),
83 Name(String),
85 Numeric(Numeric),
87 Oid(u32),
89 Record(Vec<Option<Value>>),
91 Time(NaiveTime),
93 Timestamp(CheckedTimestamp<NaiveDateTime>),
95 TimestampTz(CheckedTimestamp<DateTime<Utc>>),
97 Text(String),
99 BpChar(String),
101 VarChar(String),
103 Uuid(Uuid),
105 Int2Vector {
107 elements: Vec<Option<Value>>,
109 },
110 MzTimestamp(mz_repr::Timestamp),
112 Range(Range<Box<Value>>),
114 MzAclItem(MzAclItem),
117 AclItem(AclItem),
120}
121
122impl Value {
123 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 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 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 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 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 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 pub fn encode_binary(&self, ty: &Type, buf: &mut BytesMut) -> Result<(), io::Error> {
407 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 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 Err("binary encoding of list types is not implemented".into())
484 }
485 Value::Map(_) => {
486 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 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, 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, ScalarType::List { .. } => false, ScalarType::Map { .. } => false, 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 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 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 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 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 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 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
890pub 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 #[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}