mysql_common/binlog/
jsonb.rs

1// Copyright (c) 2021 Anatoly Ikorsky
2//
3// Licensed under the Apache License, Version 2.0
4// <LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0> or the MIT
5// license <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
6// option. All files in the project carrying such notice may not be copied,
7// modified, or distributed except according to those terms.
8
9//! MySql internal binary JSON representation.
10
11use std::{
12    borrow::Cow,
13    collections::BTreeMap,
14    convert::{TryFrom, TryInto},
15    fmt, io,
16    iter::FromIterator,
17    marker::PhantomData,
18    str::{Utf8Error, from_utf8},
19};
20
21use base64::{Engine, prelude::BASE64_STANDARD};
22use serde_json::Number;
23
24use crate::{
25    constants::ColumnType,
26    io::ParseBuf,
27    misc::{
28        raw::{Const, RawBytes, bytes::BareU16Bytes, int::*},
29        unexpected_buf_eof,
30    },
31    proto::{MyDeserialize, MySerialize},
32};
33
34use super::{decimal::Decimal, time::MysqlTime};
35
36impl fmt::Debug for Value<'_> {
37    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
38        match self {
39            Value::Null => write!(f, "null"),
40            Value::Bool(x) => write!(f, "{}", x),
41            Value::I16(x) => write!(f, "{}", x),
42            Value::U16(x) => write!(f, "{}", x),
43            Value::I32(x) => write!(f, "{}", x),
44            Value::U32(x) => write!(f, "{}", x),
45            Value::I64(x) => write!(f, "{}", x),
46            Value::U64(x) => write!(f, "{}", x),
47            Value::F64(x) => write!(f, "{}", x),
48            Value::String(x) => write!(f, "{:?}", x),
49            Value::SmallArray(a) => f.debug_list().entries(a.iter()).finish(),
50            Value::LargeArray(a) => f.debug_list().entries(a.iter()).finish(),
51            Value::SmallObject(o) => f
52                .debug_map()
53                .entries(o.iter().filter_map(|x| x.ok()))
54                .finish(),
55            Value::LargeObject(o) => f
56                .debug_map()
57                .entries(o.iter().filter_map(|x| x.ok()))
58                .finish(),
59            Value::Opaque(x) => write!(f, "<{:?}>", x),
60        }
61    }
62}
63
64/// An iterator over keys of an object.
65#[derive(Copy, Clone, PartialEq)]
66pub struct ArrayIter<'a, T> {
67    cur: u32,
68    arr: &'a ComplexValue<'a, T, Array>,
69}
70
71impl<'a, T: StorageFormat> Iterator for ArrayIter<'a, T> {
72    type Item = io::Result<Value<'a>>;
73
74    fn next(&mut self) -> Option<Self::Item> {
75        match self.arr.elem_at(self.cur).transpose() {
76            Some(x) => {
77                self.cur += 1;
78                Some(x)
79            }
80            None => None,
81        }
82    }
83}
84
85/// A key of a JSONB object.
86#[derive(Debug, Clone, Eq, PartialEq, Hash)]
87#[repr(transparent)]
88pub struct ObjectKey<'a>(RawBytes<'a, BareU16Bytes>);
89
90impl<'a> ObjectKey<'a> {
91    pub fn new(key: impl Into<Cow<'a, [u8]>>) -> Self {
92        Self(RawBytes::new(key))
93    }
94
95    /// Returns the raw value of a key.
96    pub fn value_raw(&'a self) -> &'a [u8] {
97        self.0.as_bytes()
98    }
99
100    /// Returns the value of a key as a string (lossy converted).
101    pub fn value(&'a self) -> Cow<'a, str> {
102        self.0.as_str()
103    }
104}
105
106/// A key of a JSONB object.
107#[derive(Clone, Eq, PartialEq, Hash)]
108#[repr(transparent)]
109pub struct JsonbString<'a>(RawBytes<'a, VarLen>);
110
111impl<'a> JsonbString<'a> {
112    pub fn new(string: impl Into<Cow<'a, [u8]>>) -> Self {
113        Self(RawBytes::new(string))
114    }
115
116    /// Returns the raw string value.
117    pub fn str_raw(&'a self) -> &'a [u8] {
118        self.0.as_bytes()
119    }
120
121    /// Returns the string value as UTF-8 string (lossy converted).
122    pub fn str(&'a self) -> Cow<'a, str> {
123        self.0.as_str()
124    }
125
126    pub fn into_owned(self) -> JsonbString<'static> {
127        JsonbString(self.0.into_owned())
128    }
129
130    pub fn into_raw(self) -> Cow<'a, [u8]> {
131        self.0.0
132    }
133}
134
135impl fmt::Debug for JsonbString<'_> {
136    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
137        write!(f, "{}", self.str())
138    }
139}
140
141impl<'de> MyDeserialize<'de> for JsonbString<'de> {
142    const SIZE: Option<usize> = None;
143    type Ctx = ();
144
145    fn deserialize((): Self::Ctx, buf: &mut ParseBuf<'de>) -> io::Result<Self> {
146        Ok(Self(buf.parse(())?))
147    }
148}
149
150impl MySerialize for JsonbString<'_> {
151    fn serialize(&self, buf: &mut Vec<u8>) {
152        self.0.serialize(buf);
153    }
154}
155
156/// An iterator over entries of an object.
157#[derive(Copy, Clone, PartialEq)]
158pub struct ObjectIter<'a, T> {
159    cur: u32,
160    obj: &'a ComplexValue<'a, T, Object>,
161}
162
163impl<'a, T: StorageFormat> Iterator for ObjectIter<'a, T> {
164    type Item = io::Result<(ObjectKey<'a>, Value<'a>)>;
165
166    fn next(&mut self) -> Option<Self::Item> {
167        let key = self.obj.key_at(self.cur).transpose();
168        let val = self.obj.elem_at(self.cur).transpose();
169        match key {
170            Some(key) => match val {
171                Some(val) => {
172                    self.cur += 1;
173                    Some(key.and_then(|key| val.map(|val| (key, val))))
174                }
175                None => None,
176            },
177            None => None,
178        }
179    }
180}
181
182/// An iterator over keys of an object.
183#[derive(Copy, Clone, PartialEq)]
184pub struct ObjectKeys<'a, T> {
185    cur: u32,
186    obj: &'a ComplexValue<'a, T, Object>,
187}
188
189impl<'a, T: StorageFormat> Iterator for ObjectKeys<'a, T> {
190    type Item = io::Result<ObjectKey<'a>>;
191
192    fn next(&mut self) -> Option<Self::Item> {
193        match self.obj.key_at(self.cur).transpose() {
194            Some(x) => {
195                self.cur += 1;
196                Some(x)
197            }
198            None => None,
199        }
200    }
201}
202
203/// JSONB array or object.
204#[derive(Clone, PartialEq)]
205pub struct ComplexValue<'a, T, U> {
206    element_count: u32,
207    data: Cow<'a, [u8]>,
208    __phantom: PhantomData<(T, U)>,
209}
210
211impl<'a, T, U> ComplexValue<'a, T, U> {
212    /// Returns a `'static` version of `self`.
213    pub fn into_owned(self) -> ComplexValue<'static, T, U> {
214        ComplexValue {
215            element_count: self.element_count,
216            data: Cow::Owned(self.data.into_owned()),
217            __phantom: PhantomData,
218        }
219    }
220
221    /// Returns the number of elements.
222    pub fn element_count(&self) -> u32 {
223        self.element_count
224    }
225}
226
227#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
228pub enum JsonbError {
229    #[error("Malformed jsonb: invalid key offset pointer.")]
230    InvalidKeyOffsetPointer,
231    #[error("Malformed jsonb: invalid value offset pointer.")]
232    InvalidValueOffsetPointer,
233    #[error("Malformed jsonb: jsonb array/object size is larger than the whole array/object.")]
234    SizeOverflow,
235}
236
237impl<'a, T: StorageFormat> ComplexValue<'a, T, Object> {
238    /// Returns a key with the given index.
239    ///
240    /// Returns `None` if `pos >= self.element_count()`.
241    pub fn key_at(&'a self, pos: u32) -> io::Result<Option<ObjectKey<'a>>> {
242        if pos >= self.element_count {
243            return Ok(None);
244        }
245
246        let entry_offset = 2 * T::OFFSET_SIZE as usize + T::KEY_ENTRY_SIZE as usize * pos as usize;
247
248        let mut entry_buf = ParseBuf(&self.data[entry_offset..]);
249        let key_offset = T::eat_offset(&mut entry_buf) as usize;
250        let key_length = entry_buf.eat_u16_le() as usize;
251
252        if (key_offset
253            < entry_offset
254                + ((self.element_count - pos) as usize) * T::KEY_ENTRY_SIZE as usize
255                + (self.element_count as usize) * T::VALUE_ENTRY_SIZE as usize)
256            || (self.data.len() < key_offset + key_length)
257        {
258            return Err(io::Error::new(
259                io::ErrorKind::InvalidData,
260                JsonbError::InvalidKeyOffsetPointer,
261            ));
262        }
263
264        let key_val = &self.data[key_offset..(key_offset + key_length)];
265        Ok(Some(ObjectKey::new(key_val)))
266    }
267
268    /// Returns an iterator over keys of an object.
269    pub fn keys(&'a self) -> ObjectKeys<'a, T> {
270        ObjectKeys { cur: 0, obj: self }
271    }
272
273    /// Returns an iterator over entries of an object.
274    pub fn iter(&'a self) -> ObjectIter<'a, T> {
275        ObjectIter { cur: 0, obj: self }
276    }
277}
278
279impl<'a, T: StorageFormat> ComplexValue<'a, T, Array> {
280    /// Returns an iterator over entries of an array.
281    pub fn iter(&'a self) -> ArrayIter<'a, T> {
282        ArrayIter { cur: 0, arr: self }
283    }
284}
285
286impl<'a, T: StorageFormat, U: ComplexType> ComplexValue<'a, T, U> {
287    /// Returns an element at the given position.
288    ///
289    /// * for arrays  — returns an element at the given position,
290    /// * for objects — returns an element with the given key index.
291    ///
292    /// Returns `None` if `pos >= self.element_count()`.
293    pub fn elem_at(&'a self, pos: u32) -> io::Result<Option<Value<'a>>> {
294        if pos >= self.element_count {
295            return Ok(None);
296        }
297
298        let entry_size = T::VALUE_ENTRY_SIZE as usize;
299        let entry_offset = U::value_entry_offset::<T>(self.element_count, pos);
300
301        let value_type = JsonbType::try_from(self.data[entry_offset])
302            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
303
304        if value_type.is_inlined::<T>() {
305            let start = entry_offset + 1;
306            let end = start + entry_size - 1;
307            return Value::deserialize_simple(value_type, &mut ParseBuf(&self.data[start..end]))
308                .map(Some);
309        }
310
311        let value_offset = T::eat_offset(&mut ParseBuf(&self.data[(entry_offset + 1)..])) as usize;
312
313        if self.data.len() < value_offset || value_offset < entry_offset + entry_size {
314            return Err(io::Error::new(
315                io::ErrorKind::InvalidData,
316                JsonbError::InvalidValueOffsetPointer,
317            ));
318        }
319
320        Value::deserialize_value(value_type, &mut ParseBuf(&self.data[value_offset..])).map(Some)
321    }
322}
323
324impl<'a, T: StorageFormat> TryFrom<ComplexValue<'a, T, Array>> for serde_json::Value {
325    type Error = JsonbToJsonError;
326
327    fn try_from(value: ComplexValue<'a, T, Array>) -> Result<Self, Self::Error> {
328        let values = value
329            .iter()
330            .map(|x| {
331                x.map_err(Self::Error::InvalidJsonb)
332                    .and_then(TryFrom::try_from)
333            })
334            .collect::<Result<Vec<_>, _>>()?;
335        Ok(serde_json::Value::Array(values))
336    }
337}
338
339impl<'a, T: StorageFormat> TryFrom<ComplexValue<'a, T, Object>> for serde_json::Value {
340    type Error = JsonbToJsonError;
341
342    fn try_from(value: ComplexValue<'a, T, Object>) -> Result<Self, Self::Error> {
343        let k_vs = value
344            .iter()
345            .map(|x| {
346                x.map_err(Self::Error::InvalidJsonb)
347                    .and_then(|(k, v)| Ok((from_utf8(k.value_raw())?.to_owned(), v.try_into()?)))
348            })
349            .collect::<Result<serde_json::Map<_, _>, _>>()?;
350        Ok(serde_json::Value::Object(k_vs))
351    }
352}
353
354impl<'de, T: StorageFormat, U: ComplexType> MyDeserialize<'de> for ComplexValue<'de, T, U> {
355    const SIZE: Option<usize> = None;
356    type Ctx = ();
357
358    fn deserialize((): Self::Ctx, whole_buf: &mut ParseBuf<'de>) -> io::Result<Self> {
359        let mut sbuf = ParseBuf(whole_buf.0)
360            .checked_eat_buf((T::OFFSET_SIZE * 2) as usize)
361            .ok_or_else(unexpected_buf_eof)?;
362
363        let element_count = T::eat_offset(&mut sbuf);
364        let bytes = T::eat_offset(&mut sbuf);
365
366        if bytes as usize > whole_buf.len() {
367            return Err(io::Error::new(
368                io::ErrorKind::InvalidData,
369                JsonbError::SizeOverflow,
370            ));
371        }
372
373        let header_size = U::header_size::<T>(element_count);
374
375        if header_size > bytes as usize {
376            return Err(io::Error::new(
377                io::ErrorKind::InvalidData,
378                JsonbError::SizeOverflow,
379            ));
380        }
381
382        let data = whole_buf
383            .checked_eat(bytes as usize)
384            .ok_or_else(unexpected_buf_eof)?;
385
386        Ok(Self {
387            element_count,
388            data: Cow::Borrowed(data),
389            __phantom: PhantomData,
390        })
391    }
392}
393
394/// JSONB opaque value
395#[derive(Debug, Clone, PartialEq)]
396pub struct OpaqueValue<'a> {
397    value_type: Const<ColumnType, u8>,
398    data: RawBytes<'a, VarLen>,
399}
400
401impl<'a> OpaqueValue<'a> {
402    pub fn new(value_type: ColumnType, data: impl Into<Cow<'a, [u8]>>) -> Self {
403        Self {
404            value_type: Const::new(value_type),
405            data: RawBytes::new(data),
406        }
407    }
408
409    /// Returns the value type.
410    pub fn value_type(&self) -> ColumnType {
411        self.value_type.0
412    }
413
414    /// Returns the raw value data.
415    pub fn data_raw(&'a self) -> &'a [u8] {
416        self.data.as_bytes()
417    }
418
419    /// Returns the value data as a string (lossy converted).
420    pub fn data(&'a self) -> Cow<'a, str> {
421        self.data.as_str()
422    }
423
424    pub fn into_owned(self) -> OpaqueValue<'static> {
425        OpaqueValue {
426            value_type: self.value_type,
427            data: self.data.into_owned(),
428        }
429    }
430
431    pub fn into_data(self) -> Cow<'a, [u8]> {
432        self.data.0
433    }
434}
435
436/// Structured in-memory representation of a JSON value.
437///
438/// You can get this value using [`Value::parse`].
439///
440/// You can convert this into a [`serde_json::Value`] (using [`From`] impl). Opaque values will
441/// be handled as follows:
442///
443/// *   [`ColumnType::MYSQL_TYPE_NEWDECIMAL`] — will be converted to string
444/// *   [`ColumnType::MYSQL_TYPE_DATE`] — will be converted to 'YYYY-MM-DD' string
445/// *   [`ColumnType::MYSQL_TYPE_TIME`] — will be converted to `[-][h]hh:mm::ss.µµµµµµ` string
446/// *   [`ColumnType::MYSQL_TYPE_DATETIME`] and [`ColumnType::MYSQL_TYPE_TIMESTAMP`]
447///     — will be converted to `YYYY-MM-DD hh:mm::ss.µµµµµµ` string
448/// *   other opaque values will be represented as strings in the form `base64:type<id>:<data>`
449///     where:
450///     -   `<id>` — [`ColumnType`] integer value
451///     -   `<data>` — base64-encoded opaque data
452#[derive(Debug, Clone, PartialEq)]
453pub enum JsonDom {
454    Container(JsonContainer),
455    Scalar(JsonScalar),
456}
457
458impl From<JsonDom> for serde_json::Value {
459    fn from(value: JsonDom) -> Self {
460        match value {
461            JsonDom::Container(json_container) => json_container.into(),
462            JsonDom::Scalar(json_scalar) => json_scalar.into(),
463        }
464    }
465}
466
467/// [`JsonDom`] container.
468#[derive(Debug, Clone, PartialEq)]
469pub enum JsonContainer {
470    Array(Vec<JsonDom>),
471    Object(BTreeMap<String, JsonDom>),
472}
473
474impl From<JsonContainer> for serde_json::Value {
475    fn from(value: JsonContainer) -> Self {
476        match value {
477            JsonContainer::Array(vec) => {
478                serde_json::Value::Array(Vec::from_iter(vec.into_iter().map(|x| x.into())))
479            }
480            JsonContainer::Object(btree_map) => serde_json::Value::Object(
481                serde_json::Map::from_iter(btree_map.into_iter().map(|(k, v)| (k, v.into()))),
482            ),
483        }
484    }
485}
486
487/// [`JsonDom`] scalar value.
488#[derive(Debug, Clone, PartialEq)]
489pub enum JsonScalar {
490    Boolean(bool),
491    DateTime(MysqlTime),
492    Null,
493    Number(JsonNumber),
494    Opaque(JsonOpaque),
495    String(String),
496}
497
498impl From<JsonScalar> for serde_json::Value {
499    fn from(value: JsonScalar) -> Self {
500        match value {
501            JsonScalar::Boolean(x) => serde_json::Value::Bool(x),
502            JsonScalar::DateTime(mysql_time) => {
503                serde_json::Value::String(format!("{:.6}", mysql_time))
504            }
505            JsonScalar::Null => serde_json::Value::Null,
506            JsonScalar::Number(json_number) => json_number.into(),
507            JsonScalar::Opaque(json_opaque) => json_opaque.into(),
508            JsonScalar::String(x) => serde_json::Value::String(x),
509        }
510    }
511}
512
513/// [`JsonDom`] number.
514#[derive(Debug, Clone, PartialEq)]
515pub enum JsonNumber {
516    Decimal(Decimal),
517    Double(f64),
518    Int(i64),
519    Uint(u64),
520}
521
522impl From<JsonNumber> for serde_json::Value {
523    fn from(value: JsonNumber) -> Self {
524        match value {
525            JsonNumber::Decimal(decimal) => serde_json::Value::String(decimal.to_string()),
526            JsonNumber::Double(x) => serde_json::Value::Number(
527                Number::from_f64(x)
528                    // infinities an NaN are rendered as `0`
529                    .unwrap_or_else(|| Number::from(0_u64)),
530            ),
531            JsonNumber::Int(x) => serde_json::Value::Number(Number::from(x)),
532            JsonNumber::Uint(x) => serde_json::Value::Number(Number::from(x)),
533        }
534    }
535}
536
537/// [`JsonDom`] opaque value.
538#[derive(Debug, Clone, PartialEq)]
539pub struct JsonOpaque {
540    field_type: ColumnType,
541    value: Vec<u8>,
542}
543
544impl fmt::Display for JsonOpaque {
545    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
546        write!(
547            f,
548            "base64:type{}:{}",
549            self.field_type as u8,
550            BASE64_STANDARD.encode(&self.value)
551        )
552    }
553}
554
555impl From<JsonOpaque> for serde_json::Value {
556    fn from(value: JsonOpaque) -> Self {
557        serde_json::Value::String(value.to_string())
558    }
559}
560
561/// Deserialized Jsonb value.
562///
563/// You can [`Value::parse`] it to a structured [`JsonDom`] value.
564#[derive(Clone, PartialEq)]
565pub enum Value<'a> {
566    Null,
567    Bool(bool),
568    I16(i16),
569    U16(u16),
570    I32(i32),
571    U32(u32),
572    I64(i64),
573    U64(u64),
574    F64(f64),
575    String(JsonbString<'a>),
576    SmallArray(ComplexValue<'a, Small, Array>),
577    LargeArray(ComplexValue<'a, Large, Array>),
578    SmallObject(ComplexValue<'a, Small, Object>),
579    LargeObject(ComplexValue<'a, Large, Object>),
580    Opaque(OpaqueValue<'a>),
581}
582
583impl<'a> Value<'a> {
584    /// Returns a `'static` version of `self`.
585    pub fn into_owned(self) -> Value<'static> {
586        match self {
587            Value::Null => Value::Null,
588            Value::Bool(x) => Value::Bool(x),
589            Value::I16(x) => Value::I16(x),
590            Value::U16(x) => Value::U16(x),
591            Value::I32(x) => Value::I32(x),
592            Value::U32(x) => Value::U32(x),
593            Value::I64(x) => Value::I64(x),
594            Value::U64(x) => Value::U64(x),
595            Value::F64(x) => Value::F64(x),
596            Value::String(x) => Value::String(x.into_owned()),
597            Value::SmallArray(x) => Value::SmallArray(x.into_owned()),
598            Value::LargeArray(x) => Value::LargeArray(x.into_owned()),
599            Value::SmallObject(x) => Value::SmallObject(x.into_owned()),
600            Value::LargeObject(x) => Value::LargeObject(x.into_owned()),
601            Value::Opaque(x) => Value::Opaque(x.into_owned()),
602        }
603    }
604
605    fn deserialize_value(value_type: JsonbType, buf: &mut ParseBuf<'a>) -> io::Result<Self> {
606        match value_type {
607            JsonbType::JSONB_TYPE_SMALL_OBJECT => Ok(Value::SmallObject(buf.parse(())?)),
608            JsonbType::JSONB_TYPE_LARGE_OBJECT => Ok(Value::LargeObject(buf.parse(())?)),
609            JsonbType::JSONB_TYPE_SMALL_ARRAY => Ok(Value::SmallArray(buf.parse(())?)),
610            JsonbType::JSONB_TYPE_LARGE_ARRAY => Ok(Value::LargeArray(buf.parse(())?)),
611            _ => Value::deserialize_simple(value_type, buf),
612        }
613    }
614
615    fn deserialize_simple(value_type: JsonbType, buf: &mut ParseBuf<'a>) -> io::Result<Self> {
616        match value_type {
617            JsonbType::JSONB_TYPE_LITERAL => Value::deserialize_literal(buf),
618            JsonbType::JSONB_TYPE_INT16 => buf.parse::<RawInt<LeI16>>(()).map(|x| Value::I16(*x)),
619            JsonbType::JSONB_TYPE_UINT16 => buf.parse::<RawInt<LeU16>>(()).map(|x| Value::U16(*x)),
620            JsonbType::JSONB_TYPE_INT32 => buf.parse::<RawInt<LeI32>>(()).map(|x| Value::I32(*x)),
621            JsonbType::JSONB_TYPE_UINT32 => buf.parse::<RawInt<LeU32>>(()).map(|x| Value::U32(*x)),
622            JsonbType::JSONB_TYPE_INT64 => buf.parse::<RawInt<LeI64>>(()).map(|x| Value::I64(*x)),
623            JsonbType::JSONB_TYPE_UINT64 => buf.parse::<RawInt<LeU64>>(()).map(|x| Value::U64(*x)),
624            JsonbType::JSONB_TYPE_DOUBLE => buf.parse(()).map(Value::F64),
625            JsonbType::JSONB_TYPE_STRING => Value::deserialize_string(buf),
626            JsonbType::JSONB_TYPE_OPAQUE => Value::deserialize_opaque(buf),
627            JsonbType::JSONB_TYPE_SMALL_OBJECT
628            | JsonbType::JSONB_TYPE_LARGE_OBJECT
629            | JsonbType::JSONB_TYPE_SMALL_ARRAY
630            | JsonbType::JSONB_TYPE_LARGE_ARRAY => unreachable!(),
631        }
632    }
633
634    fn deserialize_literal(buf: &mut ParseBuf<'a>) -> io::Result<Self> {
635        let literal_type = buf.checked_eat_u8().ok_or_else(unexpected_buf_eof)?;
636
637        match LiteralType::try_from(literal_type)
638            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?
639        {
640            LiteralType::JSONB_NULL_LITERAL => Ok(Value::Null),
641            LiteralType::JSONB_TRUE_LITERAL => Ok(Value::Bool(true)),
642            LiteralType::JSONB_FALSE_LITERAL => Ok(Value::Bool(false)),
643        }
644    }
645
646    fn deserialize_string(buf: &mut ParseBuf<'a>) -> io::Result<Self> {
647        buf.parse(()).map(Value::String)
648    }
649
650    fn deserialize_opaque(buf: &mut ParseBuf<'a>) -> io::Result<Self> {
651        Ok(Value::Opaque(OpaqueValue {
652            value_type: buf.parse(())?,
653            data: buf.parse(())?,
654        }))
655    }
656
657    /// Returns true if this value is an array.
658    pub fn is_array(&self) -> bool {
659        matches!(self, Value::SmallArray { .. } | Value::LargeArray { .. })
660    }
661
662    /// Returns true if this value is an object.
663    pub fn is_object(&self) -> bool {
664        matches!(self, Value::SmallObject { .. } | Value::LargeObject { .. })
665    }
666
667    /// Returns true if this value is an int (i16, i32 or i64).
668    pub fn is_int(&self) -> bool {
669        matches!(self, Value::I16(_) | Value::I32(_) | Value::I64(_))
670    }
671
672    /// Returns true if this value is an int (u16, u32 or u64).
673    pub fn is_uint(&self) -> bool {
674        matches!(self, Value::U16(_) | Value::U32(_) | Value::U64(_))
675    }
676
677    /// Returns true if this value is f64.
678    pub fn is_double(&self) -> bool {
679        matches!(self, Value::F64(_))
680    }
681
682    /// Returns the number of elements in array or object.
683    ///
684    /// Returns `None` on none-array/non-object values.
685    pub fn element_count(&self) -> Option<u32> {
686        match self {
687            Value::SmallArray(x) => Some(x.element_count()),
688            Value::LargeArray(x) => Some(x.element_count()),
689            Value::SmallObject(x) => Some(x.element_count()),
690            Value::LargeObject(x) => Some(x.element_count()),
691            _ => None,
692        }
693    }
694
695    /// Returns the field type of an opaque value.
696    ///
697    /// Returns `None` on non-opaque values.
698    pub fn field_type(&self) -> Option<ColumnType> {
699        match self {
700            Value::Opaque(OpaqueValue { value_type, .. }) => Some(**value_type),
701            _ => None,
702        }
703    }
704
705    /// Parse this value to a structured representation.
706    pub fn parse(self) -> io::Result<JsonDom> {
707        match self {
708            Value::Null => Ok(JsonDom::Scalar(JsonScalar::Null)),
709            Value::Bool(value) => Ok(JsonDom::Scalar(JsonScalar::Boolean(value))),
710            Value::I16(x) => Ok(JsonDom::Scalar(JsonScalar::Number(JsonNumber::Int(
711                x as i64,
712            )))),
713            Value::U16(x) => Ok(JsonDom::Scalar(JsonScalar::Number(JsonNumber::Uint(
714                x as u64,
715            )))),
716            Value::I32(x) => Ok(JsonDom::Scalar(JsonScalar::Number(JsonNumber::Int(
717                x as i64,
718            )))),
719            Value::U32(x) => Ok(JsonDom::Scalar(JsonScalar::Number(JsonNumber::Uint(
720                x as u64,
721            )))),
722            Value::I64(x) => Ok(JsonDom::Scalar(JsonScalar::Number(JsonNumber::Int(x)))),
723            Value::U64(x) => Ok(JsonDom::Scalar(JsonScalar::Number(JsonNumber::Uint(x)))),
724            Value::F64(x) => Ok(JsonDom::Scalar(JsonScalar::Number(JsonNumber::Double(x)))),
725            Value::String(jsonb_string) => {
726                let s = match jsonb_string.into_raw() {
727                    Cow::Borrowed(x) => Cow::Borrowed(
728                        std::str::from_utf8(x)
729                            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?,
730                    ),
731                    Cow::Owned(x) => Cow::Owned(
732                        String::from_utf8(x)
733                            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?,
734                    ),
735                };
736                Ok(JsonDom::Scalar(JsonScalar::String(s.into_owned())))
737            }
738            Value::SmallArray(complex_value) => {
739                let mut elements = Vec::with_capacity(complex_value.element_count() as usize);
740                for i in 0.. {
741                    if let Some(value) = complex_value.elem_at(i)? {
742                        let y = value.parse()?;
743                        elements.push(y);
744                    } else {
745                        break;
746                    }
747                }
748                Ok(JsonDom::Container(JsonContainer::Array(elements)))
749            }
750            Value::LargeArray(complex_value) => {
751                let mut elements = Vec::with_capacity(complex_value.element_count() as usize);
752                for value in complex_value.iter() {
753                    elements.push(value?.parse()?);
754                }
755                Ok(JsonDom::Container(JsonContainer::Array(elements)))
756            }
757            Value::SmallObject(complex_value) => {
758                let mut elements = BTreeMap::new();
759                for value in complex_value.iter() {
760                    let (key, value) = value?;
761                    elements.insert(key.value().into_owned(), value.parse()?);
762                }
763                Ok(JsonDom::Container(JsonContainer::Object(elements)))
764            }
765            Value::LargeObject(complex_value) => {
766                let mut elements = BTreeMap::new();
767                for value in complex_value.iter() {
768                    let (key, value) = value?;
769                    elements.insert(key.value().into_owned(), value.parse()?);
770                }
771                Ok(JsonDom::Container(JsonContainer::Object(elements)))
772            }
773            Value::Opaque(opaque_value) => match opaque_value.value_type() {
774                ColumnType::MYSQL_TYPE_NEWDECIMAL => {
775                    let data = opaque_value.data_raw();
776
777                    Ok(JsonDom::Scalar(JsonScalar::Number(JsonNumber::Decimal(
778                        Decimal::read_packed(data, false)?,
779                    ))))
780                }
781                ColumnType::MYSQL_TYPE_DATE => {
782                    let packed_value =
783                        opaque_value.data_raw().first_chunk::<8>().ok_or_else(|| {
784                            io::Error::new(
785                                io::ErrorKind::InvalidData,
786                                "not enough data to decode MYSQL_TYPE_DATE",
787                            )
788                        })?;
789                    let packed_value = i64::from_le_bytes(*packed_value);
790                    Ok(JsonDom::Scalar(JsonScalar::DateTime(
791                        MysqlTime::from_int64_date_packed(packed_value),
792                    )))
793                }
794                ColumnType::MYSQL_TYPE_TIME => {
795                    let packed_value =
796                        opaque_value.data_raw().first_chunk::<8>().ok_or_else(|| {
797                            io::Error::new(
798                                io::ErrorKind::InvalidData,
799                                "not enough data to decode MYSQL_TYPE_TIME",
800                            )
801                        })?;
802                    let packed_value = i64::from_le_bytes(*packed_value);
803                    Ok(JsonDom::Scalar(JsonScalar::DateTime(
804                        MysqlTime::from_int64_time_packed(packed_value),
805                    )))
806                }
807                ColumnType::MYSQL_TYPE_DATETIME | ColumnType::MYSQL_TYPE_TIMESTAMP => {
808                    let packed_value =
809                        opaque_value.data_raw().first_chunk::<8>().ok_or_else(|| {
810                            io::Error::new(
811                                io::ErrorKind::InvalidData,
812                                "not enough data to decode MYSQL_TYPE_DATETIME",
813                            )
814                        })?;
815                    let packed_value = i64::from_le_bytes(*packed_value);
816                    Ok(JsonDom::Scalar(JsonScalar::DateTime(
817                        MysqlTime::from_int64_datetime_packed(packed_value),
818                    )))
819                }
820
821                field_type => Ok(JsonDom::Scalar(JsonScalar::Opaque(JsonOpaque {
822                    field_type,
823                    value: opaque_value.data.0.into_owned(),
824                }))),
825            },
826        }
827    }
828}
829
830impl<'a> TryFrom<Value<'a>> for serde_json::Value {
831    type Error = JsonbToJsonError;
832
833    fn try_from(value: Value<'a>) -> Result<Self, Self::Error> {
834        match value {
835            Value::Null => Ok(serde_json::Value::Null),
836            Value::Bool(x) => Ok(serde_json::Value::Bool(x)),
837            Value::I16(x) => Ok(x.into()),
838            Value::U16(x) => Ok(x.into()),
839            Value::I32(x) => Ok(x.into()),
840            Value::U32(x) => Ok(x.into()),
841            Value::I64(x) => Ok(x.into()),
842            Value::U64(x) => Ok(x.into()),
843            Value::F64(x) => Ok(x.into()),
844            Value::String(s) => Ok(from_utf8(s.str_raw())?.into()),
845            Value::SmallArray(x) => x.try_into(),
846            Value::LargeArray(x) => x.try_into(),
847            Value::SmallObject(x) => x.try_into(),
848            Value::LargeObject(x) => x.try_into(),
849            Value::Opaque(_) => Err(Self::Error::Opaque),
850        }
851    }
852}
853
854#[derive(Debug, thiserror::Error)]
855pub enum JsonbToJsonError {
856    #[error("JSONB value is invalid: {}", _0)]
857    InvalidJsonb(#[from] io::Error),
858    #[error("JSONB contains an opaque value")]
859    Opaque,
860    #[error("JSONB contains invalid UTF-8 char sequences: {}", _0)]
861    InvalidUtf8(#[from] Utf8Error),
862}
863
864impl<'de> MyDeserialize<'de> for Value<'de> {
865    const SIZE: Option<usize> = None;
866    type Ctx = ();
867
868    fn deserialize((): Self::Ctx, buf: &mut ParseBuf<'de>) -> io::Result<Self> {
869        if buf.is_empty() {
870            // We'll interpret an empty jsonb value as null (MySql does so)
871            return Ok(Value::Null);
872        }
873
874        let value_type = JsonbType::try_from(buf.eat_u8())
875            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
876
877        Value::deserialize_value(value_type, buf)
878    }
879}
880
881/// Type of a complex jsonb value (array or object).
882pub trait ComplexType {
883    const IS_ARRAY: bool;
884
885    /// Calculates header size for this type.
886    fn header_size<T: StorageFormat>(element_count: u32) -> usize;
887
888    /// Calculates an offset to the given value entry.
889    fn value_entry_offset<T: StorageFormat>(element_count: u32, pos: u32) -> usize;
890}
891
892/// An Object (see [`ComplexType`]).
893#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
894pub struct Object;
895
896impl ComplexType for Object {
897    const IS_ARRAY: bool = false;
898
899    fn header_size<T: StorageFormat>(element_count: u32) -> usize {
900        let mut header_size = 2 * T::OFFSET_SIZE as usize;
901        header_size += element_count as usize * T::KEY_ENTRY_SIZE as usize;
902        header_size += element_count as usize * T::VALUE_ENTRY_SIZE as usize;
903        header_size
904    }
905
906    fn value_entry_offset<T: StorageFormat>(element_count: u32, pos: u32) -> usize {
907        let mut first_entry_offset = 2 * T::OFFSET_SIZE as usize;
908        first_entry_offset += element_count as usize * T::KEY_ENTRY_SIZE as usize;
909        first_entry_offset + T::VALUE_ENTRY_SIZE as usize * pos as usize
910    }
911}
912
913/// An Array (see [`ComplexType`]).
914#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
915pub struct Array;
916
917impl ComplexType for Array {
918    const IS_ARRAY: bool = true;
919
920    fn header_size<T: StorageFormat>(element_count: u32) -> usize {
921        let mut header_size = 2 * T::OFFSET_SIZE as usize;
922        header_size += element_count as usize * T::VALUE_ENTRY_SIZE as usize;
923        header_size
924    }
925
926    fn value_entry_offset<T: StorageFormat>(_element_count: u32, pos: u32) -> usize {
927        let first_entry_offset = 2 * T::OFFSET_SIZE as usize;
928        first_entry_offset + T::VALUE_ENTRY_SIZE as usize * pos as usize
929    }
930}
931
932/// JSONB storage format for objects and arrays. See [`ComplexType`].
933pub trait StorageFormat {
934    const IS_LARGE: bool;
935    /// The size of offset or size fields.
936    const OFFSET_SIZE: u8;
937    const KEY_ENTRY_SIZE: u8 = Self::OFFSET_SIZE + 2;
938    const VALUE_ENTRY_SIZE: u8 = Self::OFFSET_SIZE + 1;
939
940    fn eat_offset(buf: &mut ParseBuf<'_>) -> u32;
941}
942
943/// Small array/object storage format. See [`StorageFormat`].
944#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
945pub struct Small;
946
947impl StorageFormat for Small {
948    const IS_LARGE: bool = false;
949    const OFFSET_SIZE: u8 = 2;
950
951    fn eat_offset(buf: &mut ParseBuf<'_>) -> u32 {
952        buf.eat_u16_le() as u32
953    }
954}
955
956/// Large array/object storage format. See [`StorageFormat`].
957#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
958pub struct Large;
959
960impl StorageFormat for Large {
961    const IS_LARGE: bool = true;
962    const OFFSET_SIZE: u8 = 4;
963
964    fn eat_offset(buf: &mut ParseBuf<'_>) -> u32 {
965        buf.eat_u32_le()
966    }
967}
968
969/// 1-byte JSONB type marker
970#[allow(non_camel_case_types)]
971#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
972#[repr(u8)]
973pub enum JsonbType {
974    JSONB_TYPE_SMALL_OBJECT = 0x0,
975    JSONB_TYPE_LARGE_OBJECT = 0x1,
976    JSONB_TYPE_SMALL_ARRAY = 0x2,
977    JSONB_TYPE_LARGE_ARRAY = 0x3,
978    JSONB_TYPE_LITERAL = 0x4,
979    JSONB_TYPE_INT16 = 0x5,
980    JSONB_TYPE_UINT16 = 0x6,
981    JSONB_TYPE_INT32 = 0x7,
982    JSONB_TYPE_UINT32 = 0x8,
983    JSONB_TYPE_INT64 = 0x9,
984    JSONB_TYPE_UINT64 = 0xA,
985    JSONB_TYPE_DOUBLE = 0xB,
986    JSONB_TYPE_STRING = 0xC,
987    JSONB_TYPE_OPAQUE = 0xF,
988}
989
990impl JsonbType {
991    fn is_inlined<T: StorageFormat>(&self) -> bool {
992        match self {
993            JsonbType::JSONB_TYPE_LITERAL
994            | JsonbType::JSONB_TYPE_INT16
995            | JsonbType::JSONB_TYPE_UINT16 => true,
996            JsonbType::JSONB_TYPE_INT32 | JsonbType::JSONB_TYPE_UINT32 => T::IS_LARGE,
997            _ => false,
998        }
999    }
1000}
1001
1002#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, thiserror::Error)]
1003#[error("Unknown JSONB type {}", _0)]
1004#[repr(transparent)]
1005pub struct UnknownJsonbType(pub u8);
1006
1007impl From<UnknownJsonbType> for u8 {
1008    fn from(x: UnknownJsonbType) -> Self {
1009        x.0
1010    }
1011}
1012
1013impl TryFrom<u8> for JsonbType {
1014    type Error = UnknownJsonbType;
1015
1016    fn try_from(value: u8) -> Result<Self, Self::Error> {
1017        match value {
1018            0x0 => Ok(Self::JSONB_TYPE_SMALL_OBJECT),
1019            0x1 => Ok(Self::JSONB_TYPE_LARGE_OBJECT),
1020            0x2 => Ok(Self::JSONB_TYPE_SMALL_ARRAY),
1021            0x3 => Ok(Self::JSONB_TYPE_LARGE_ARRAY),
1022            0x4 => Ok(Self::JSONB_TYPE_LITERAL),
1023            0x5 => Ok(Self::JSONB_TYPE_INT16),
1024            0x6 => Ok(Self::JSONB_TYPE_UINT16),
1025            0x7 => Ok(Self::JSONB_TYPE_INT32),
1026            0x8 => Ok(Self::JSONB_TYPE_UINT32),
1027            0x9 => Ok(Self::JSONB_TYPE_INT64),
1028            0xA => Ok(Self::JSONB_TYPE_UINT64),
1029            0xB => Ok(Self::JSONB_TYPE_DOUBLE),
1030            0xC => Ok(Self::JSONB_TYPE_STRING),
1031            0xF => Ok(Self::JSONB_TYPE_OPAQUE),
1032            x => Err(UnknownJsonbType(x)),
1033        }
1034    }
1035}
1036
1037/// 1-byte JSONB literal type
1038#[allow(non_camel_case_types)]
1039#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
1040#[repr(u8)]
1041pub enum LiteralType {
1042    JSONB_NULL_LITERAL = 0x0,
1043    JSONB_TRUE_LITERAL = 0x1,
1044    JSONB_FALSE_LITERAL = 0x2,
1045}
1046
1047#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, thiserror::Error)]
1048#[error("Unknown JSONB literal type {}", _0)]
1049#[repr(transparent)]
1050pub struct UnknownLiteralType(pub u8);
1051
1052impl From<UnknownLiteralType> for u8 {
1053    fn from(x: UnknownLiteralType) -> Self {
1054        x.0
1055    }
1056}
1057
1058impl TryFrom<u8> for LiteralType {
1059    type Error = UnknownLiteralType;
1060
1061    fn try_from(value: u8) -> Result<Self, Self::Error> {
1062        match value {
1063            0x0 => Ok(Self::JSONB_NULL_LITERAL),
1064            0x1 => Ok(Self::JSONB_TRUE_LITERAL),
1065            0x2 => Ok(Self::JSONB_FALSE_LITERAL),
1066            x => Err(UnknownLiteralType(x)),
1067        }
1068    }
1069}
mysql_common/binlog/jsonb.rs

mysql_common/binlog/
jsonb.rs
