laser_wire/
schema.rs

1use crate::error::InvalidError;
2use crate::limits::MAX_VALUE_BYTES;
3use crate::validate::Validate;
4use serde::de::{self, SeqAccess, Visitor};
5use serde::{Deserialize, Deserializer, Serialize, Serializer};
6use sha2::{Digest, Sha256};
7use std::collections::BTreeSet;
8use std::fmt;
9
10pub const MAX_LOGICAL_SCHEMA_DEPTH: usize = 16;
11pub const MAX_LOGICAL_SCHEMA_FIELDS: usize = 4_096;
12pub const MAX_LOGICAL_SCHEMA_BYTES: usize = 1_048_576;
13pub const MAX_FIELD_NAME_BYTES: usize = 255;
14pub const MAX_FIELD_DOC_BYTES: usize = 4_096;
15pub const MAX_FIXED_BYTES: u32 = 1_048_576;
16pub const MAX_DECIMAL_PRECISION: u8 = 38;
17pub const MICROS_PER_DAY: i64 = 86_400_000_000;
18pub const PROVENANCE_FIELD_ID_START: u32 = 2_000_000_000;
19
20pub const PROVENANCE_FIELDS: &[(u32, &str)] = &[
21    (2_000_000_001, "__laser_cluster_incarnation"),
22    (2_000_000_002, "__laser_source_incarnation"),
23    (2_000_000_003, "__laser_stream_id"),
24    (2_000_000_004, "__laser_topic_id"),
25    (2_000_000_005, "__laser_partition_id"),
26    (2_000_000_006, "__laser_offset"),
27    (2_000_000_007, "__laser_row_ordinal"),
28    (2_000_000_008, "__laser_projection_id"),
29    (2_000_000_009, "__laser_projection_version"),
30    (2_000_000_010, "__laser_destination_id"),
31    (2_000_000_011, "__laser_destination_generation"),
32    (2_000_000_012, "__laser_original_payload"),
33    (2_000_000_013, "__laser_original_content_type"),
34    (2_000_000_014, "__laser_original_schema_id"),
35    (2_000_000_015, "__laser_original_headers"),
36];
37
38pub const ORIGINAL_PAYLOAD_FIELD_ID: u32 = 2_000_000_012;
39pub const ORIGINAL_PAYLOAD_FIELD_NAME: &str = "__laser_original_payload";
40pub const ORIGINAL_CONTENT_TYPE_FIELD_ID: u32 = 2_000_000_013;
41pub const ORIGINAL_CONTENT_TYPE_FIELD_NAME: &str = "__laser_original_content_type";
42pub const ORIGINAL_SCHEMA_ID_FIELD_ID: u32 = 2_000_000_014;
43pub const ORIGINAL_SCHEMA_ID_FIELD_NAME: &str = "__laser_original_schema_id";
44pub const ORIGINAL_HEADERS_FIELD_ID: u32 = 2_000_000_015;
45pub const ORIGINAL_HEADERS_FIELD_NAME: &str = "__laser_original_headers";
46pub const PARTITION_ID_FIELD_NAME: &str = "__laser_partition_id";
47pub const OFFSET_FIELD_NAME: &str = "__laser_offset";
48
49/// Immutable identity of one logical schema family.
50#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
51pub struct LogicalSchemaId(u128);
52
53impl LogicalSchemaId {
54    pub const fn from_u128(value: u128) -> Self {
55        Self(value)
56    }
57
58    pub const fn as_u128(self) -> u128 {
59        self.0
60    }
61
62    pub const fn to_bytes(self) -> [u8; 16] {
63        self.0.to_be_bytes()
64    }
65
66    pub const fn from_bytes(bytes: [u8; 16]) -> Self {
67        Self(u128::from_be_bytes(bytes))
68    }
69}
70
71impl Serialize for LogicalSchemaId {
72    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
73        serializer.serialize_bytes(&self.to_bytes())
74    }
75}
76
77impl<'de> Deserialize<'de> for LogicalSchemaId {
78    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
79        struct LogicalSchemaIdVisitor;
80
81        impl<'de> Visitor<'de> for LogicalSchemaIdVisitor {
82            type Value = LogicalSchemaId;
83
84            fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
85                formatter.write_str("16 big-endian logical schema id bytes")
86            }
87
88            fn visit_bytes<E: de::Error>(self, bytes: &[u8]) -> Result<Self::Value, E> {
89                let bytes = bytes
90                    .try_into()
91                    .map_err(|_| E::invalid_length(bytes.len(), &self))?;
92                Ok(LogicalSchemaId::from_bytes(bytes))
93            }
94
95            fn visit_byte_buf<E: de::Error>(self, bytes: Vec<u8>) -> Result<Self::Value, E> {
96                self.visit_bytes(&bytes)
97            }
98
99            fn visit_seq<A: SeqAccess<'de>>(
100                self,
101                mut sequence: A,
102            ) -> Result<Self::Value, A::Error> {
103                let mut bytes = [0_u8; 16];
104                for (index, byte) in bytes.iter_mut().enumerate() {
105                    *byte = sequence
106                        .next_element()?
107                        .ok_or_else(|| de::Error::invalid_length(index, &self))?;
108                }
109                if sequence.next_element::<u8>()?.is_some() {
110                    return Err(de::Error::invalid_length(17, &self));
111                }
112                Ok(LogicalSchemaId::from_bytes(bytes))
113            }
114        }
115
116        deserializer.deserialize_bytes(LogicalSchemaIdVisitor)
117    }
118}
119
120#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
121pub struct SchemaFingerprint(#[serde(with = "crate::encoding::bin_bytes")] pub Vec<u8>);
122
123impl SchemaFingerprint {
124    pub const BYTES: usize = 32;
125
126    pub fn new(bytes: [u8; Self::BYTES]) -> Self {
127        Self(bytes.to_vec())
128    }
129
130    pub fn as_bytes(&self) -> &[u8] {
131        &self.0
132    }
133}
134
135#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
136pub struct Digest32(#[serde(with = "crate::encoding::bin_bytes")] pub Vec<u8>);
137
138impl Digest32 {
139    pub const BYTES: usize = 32;
140
141    pub fn new(bytes: [u8; Self::BYTES]) -> Self {
142        Self(bytes.to_vec())
143    }
144
145    pub fn as_bytes(&self) -> &[u8] {
146        &self.0
147    }
148}
149
150impl Validate for Digest32 {
151    fn validate(&self) -> Result<(), InvalidError> {
152        if self.0.len() != Self::BYTES {
153            return Err(InvalidError::new(format!(
154                "digest must be {} bytes, got {}",
155                Self::BYTES,
156                self.0.len()
157            )));
158        }
159        Ok(())
160    }
161}
162
163impl Validate for SchemaFingerprint {
164    fn validate(&self) -> Result<(), InvalidError> {
165        if self.0.len() != Self::BYTES {
166            return Err(InvalidError::new(format!(
167                "schema fingerprint must be {} bytes, got {}",
168                Self::BYTES,
169                self.0.len()
170            )));
171        }
172        Ok(())
173    }
174}
175
176#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
177pub struct LogicalSchemaRef {
178    pub id: LogicalSchemaId,
179    pub version: u32,
180    pub fingerprint: SchemaFingerprint,
181}
182
183impl Validate for LogicalSchemaRef {
184    fn validate(&self) -> Result<(), InvalidError> {
185        if self.id.as_u128() == 0 {
186            return Err(InvalidError::new("logical schema id must be nonzero"));
187        }
188        if self.version == 0 {
189            return Err(InvalidError::new("logical schema version must be nonzero"));
190        }
191        self.fingerprint.validate()
192    }
193}
194
195#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
196pub struct LogicalSchema {
197    pub schema: LogicalSchemaRef,
198    pub fields: Vec<LogicalField>,
199}
200
201impl Validate for LogicalSchema {
202    fn validate(&self) -> Result<(), InvalidError> {
203        self.validate_shape()?;
204        let computed = self.compute_fingerprint()?;
205        if self.schema.fingerprint != computed {
206            return Err(InvalidError::new(
207                "logical schema fingerprint does not match its canonical bytes",
208            ));
209        }
210        Ok(())
211    }
212}
213
214impl LogicalSchema {
215    pub fn new(
216        id: LogicalSchemaId,
217        version: u32,
218        fields: Vec<LogicalField>,
219    ) -> Result<Self, InvalidError> {
220        let mut schema = Self {
221            schema: LogicalSchemaRef {
222                id,
223                version,
224                fingerprint: SchemaFingerprint(Vec::new()),
225            },
226            fields,
227        };
228        schema.validate_shape()?;
229        schema.schema.fingerprint = schema.compute_fingerprint()?;
230        Ok(schema)
231    }
232
233    pub fn canonical_fingerprint_bytes(&self) -> Result<Vec<u8>, InvalidError> {
234        self.validate_shape()?;
235        let mut bytes = b"AGDX-SCHEMA-V1\0".to_vec();
236        bytes.extend_from_slice(&self.schema.id.to_bytes());
237        bytes.extend_from_slice(&self.schema.version.to_be_bytes());
238        encode_fields(&mut bytes, &self.fields)?;
239        Ok(bytes)
240    }
241
242    pub fn compute_fingerprint(&self) -> Result<SchemaFingerprint, InvalidError> {
243        let digest: [u8; SchemaFingerprint::BYTES] =
244            Sha256::digest(self.canonical_fingerprint_bytes()?).into();
245        Ok(SchemaFingerprint::new(digest))
246    }
247
248    fn validate_shape(&self) -> Result<(), InvalidError> {
249        if self.schema.id.as_u128() == 0 {
250            return Err(InvalidError::new("logical schema id must be nonzero"));
251        }
252        if self.schema.version == 0 {
253            return Err(InvalidError::new("logical schema version must be nonzero"));
254        }
255        if self.fields.is_empty() {
256            return Err(InvalidError::new("logical schema must contain a field"));
257        }
258
259        let mut ids = BTreeSet::new();
260        let mut field_count = 0usize;
261        let mut encoded_size = 0usize;
262        validate_struct_fields(
263            &self.fields,
264            1,
265            &mut ids,
266            &mut field_count,
267            &mut encoded_size,
268            false,
269        )?;
270        Ok(())
271    }
272}
273
274pub(crate) fn validate_result_fields(fields: &[LogicalField]) -> Result<(), InvalidError> {
275    if fields.is_empty() {
276        return Err(InvalidError::new("query result schema must not be empty"));
277    }
278    let mut ids = BTreeSet::new();
279    let mut field_count = 0usize;
280    let mut encoded_size = 0usize;
281    validate_struct_fields(
282        fields,
283        1,
284        &mut ids,
285        &mut field_count,
286        &mut encoded_size,
287        true,
288    )
289}
290
291fn encode_fields(bytes: &mut Vec<u8>, fields: &[LogicalField]) -> Result<(), InvalidError> {
292    let count = u32::try_from(fields.len())
293        .map_err(|_| InvalidError::new("logical schema field count does not fit u32"))?;
294    bytes.extend_from_slice(&count.to_be_bytes());
295    for field in fields {
296        bytes.extend_from_slice(&field.id.to_be_bytes());
297        bytes.push(u8::from(field.required));
298        encode_text(bytes, &field.name)?;
299        match &field.doc {
300            Some(doc) => {
301                bytes.push(1);
302                encode_text(bytes, doc)?;
303            }
304            None => bytes.push(0),
305        }
306        encode_logical_type(bytes, &field.field_type)?;
307    }
308    Ok(())
309}
310
311fn encode_text(bytes: &mut Vec<u8>, value: &str) -> Result<(), InvalidError> {
312    let length = u32::try_from(value.len())
313        .map_err(|_| InvalidError::new("logical schema text length does not fit u32"))?;
314    bytes.extend_from_slice(&length.to_be_bytes());
315    bytes.extend_from_slice(value.as_bytes());
316    Ok(())
317}
318
319fn encode_logical_type(
320    bytes: &mut Vec<u8>,
321    logical_type: &LogicalType,
322) -> Result<(), InvalidError> {
323    match logical_type {
324        LogicalType::Boolean => bytes.push(0),
325        LogicalType::Int => bytes.push(1),
326        LogicalType::Long => bytes.push(2),
327        LogicalType::Float => bytes.push(3),
328        LogicalType::Double => bytes.push(4),
329        LogicalType::Decimal { precision, scale } => {
330            bytes.extend_from_slice(&[5, *precision, *scale]);
331        }
332        LogicalType::Date => bytes.push(6),
333        LogicalType::TimeMicros => bytes.push(7),
334        LogicalType::TimestampMicros => bytes.push(8),
335        LogicalType::TimestampTzMicros => bytes.push(9),
336        LogicalType::String => bytes.push(10),
337        LogicalType::Uuid => bytes.push(11),
338        LogicalType::Fixed { length } => {
339            bytes.push(12);
340            bytes.extend_from_slice(&length.to_be_bytes());
341        }
342        LogicalType::Binary => bytes.push(13),
343        LogicalType::Struct { fields } => {
344            bytes.push(14);
345            encode_fields(bytes, fields)?;
346        }
347        LogicalType::List {
348            element_id,
349            element_required,
350            element,
351        } => {
352            bytes.push(15);
353            bytes.extend_from_slice(&element_id.to_be_bytes());
354            bytes.push(u8::from(*element_required));
355            encode_logical_type(bytes, element)?;
356        }
357        LogicalType::Map {
358            key_id,
359            key,
360            value_id,
361            value_required,
362            value,
363        } => {
364            bytes.push(16);
365            bytes.extend_from_slice(&key_id.to_be_bytes());
366            encode_logical_type(bytes, key)?;
367            bytes.extend_from_slice(&value_id.to_be_bytes());
368            bytes.push(u8::from(*value_required));
369            encode_logical_type(bytes, value)?;
370        }
371        LogicalType::Variant => bytes.push(17),
372    }
373    Ok(())
374}
375
376#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
377pub struct LogicalField {
378    pub id: u32,
379    pub name: String,
380    pub required: bool,
381    pub field_type: LogicalType,
382    #[serde(default, skip_serializing_if = "Option::is_none")]
383    pub doc: Option<String>,
384}
385
386#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
387#[serde(tag = "kind", rename_all = "snake_case")]
388#[non_exhaustive]
389pub enum LogicalType {
390    Boolean,
391    Int,
392    Long,
393    Float,
394    Double,
395    Decimal {
396        precision: u8,
397        scale: u8,
398    },
399    Date,
400    TimeMicros,
401    TimestampMicros,
402    TimestampTzMicros,
403    String,
404    Uuid,
405    Fixed {
406        length: u32,
407    },
408    Binary,
409    Struct {
410        fields: Vec<LogicalField>,
411    },
412    List {
413        element_id: u32,
414        element_required: bool,
415        element: Box<LogicalType>,
416    },
417    Map {
418        key_id: u32,
419        key: Box<LogicalType>,
420        value_id: u32,
421        value_required: bool,
422        value: Box<LogicalType>,
423    },
424    /// Any canonical value, each carrying its own tag, for a field declared without a type.
425    Variant,
426}
427
428/// Non-parameterized logical type identity used in capability descriptors.
429#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
430#[serde(rename_all = "snake_case")]
431#[non_exhaustive]
432pub enum LogicalTypeKind {
433    Boolean,
434    Int,
435    Long,
436    Float,
437    Double,
438    Decimal,
439    Date,
440    TimeMicros,
441    TimestampMicros,
442    TimestampTzMicros,
443    String,
444    Uuid,
445    Fixed,
446    Binary,
447    Struct,
448    List,
449    Map,
450    Variant,
451}
452
453impl LogicalType {
454    pub const fn kind(&self) -> LogicalTypeKind {
455        match self {
456            Self::Boolean => LogicalTypeKind::Boolean,
457            Self::Int => LogicalTypeKind::Int,
458            Self::Long => LogicalTypeKind::Long,
459            Self::Float => LogicalTypeKind::Float,
460            Self::Double => LogicalTypeKind::Double,
461            Self::Decimal { .. } => LogicalTypeKind::Decimal,
462            Self::Date => LogicalTypeKind::Date,
463            Self::TimeMicros => LogicalTypeKind::TimeMicros,
464            Self::TimestampMicros => LogicalTypeKind::TimestampMicros,
465            Self::TimestampTzMicros => LogicalTypeKind::TimestampTzMicros,
466            Self::String => LogicalTypeKind::String,
467            Self::Uuid => LogicalTypeKind::Uuid,
468            Self::Fixed { .. } => LogicalTypeKind::Fixed,
469            Self::Binary => LogicalTypeKind::Binary,
470            Self::Struct { .. } => LogicalTypeKind::Struct,
471            Self::List { .. } => LogicalTypeKind::List,
472            Self::Map { .. } => LogicalTypeKind::Map,
473            Self::Variant => LogicalTypeKind::Variant,
474        }
475    }
476}
477
478impl LogicalType {
479    pub fn accepts_map_key(&self) -> bool {
480        matches!(
481            self,
482            Self::Boolean
483                | Self::Int
484                | Self::Long
485                | Self::Decimal { .. }
486                | Self::Date
487                | Self::TimeMicros
488                | Self::TimestampMicros
489                | Self::TimestampTzMicros
490                | Self::String
491                | Self::Uuid
492                | Self::Fixed { .. }
493                | Self::Binary
494        )
495    }
496}
497
498#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
499pub struct DecimalValue {
500    #[serde(with = "crate::encoding::bin_bytes")]
501    pub unscaled: Vec<u8>,
502    pub precision: u8,
503    pub scale: u8,
504}
505
506impl DecimalValue {
507    pub fn validate_canonical(&self) -> Result<(), InvalidError> {
508        validate_decimal(self)
509    }
510}
511
512#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
513pub struct UuidValue(#[serde(with = "crate::encoding::bin_bytes")] pub Vec<u8>);
514
515impl UuidValue {
516    pub const BYTES: usize = 16;
517
518    pub fn new(bytes: [u8; Self::BYTES]) -> Self {
519        Self(bytes.to_vec())
520    }
521
522    pub fn as_bytes(&self) -> &[u8] {
523        &self.0
524    }
525}
526
527impl fmt::Display for UuidValue {
528    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
529        if self.0.len() != Self::BYTES {
530            return Err(fmt::Error);
531        }
532        for (index, byte) in self.0.iter().enumerate() {
533            if matches!(index, 4 | 6 | 8 | 10) {
534                formatter.write_str("-")?;
535            }
536            write!(formatter, "{byte:02x}")?;
537        }
538        Ok(())
539    }
540}
541
542impl std::str::FromStr for UuidValue {
543    type Err = InvalidError;
544
545    fn from_str(value: &str) -> Result<Self, Self::Err> {
546        if value.len() != 36
547            || value
548                .bytes()
549                .enumerate()
550                .any(|(index, byte)| matches!(index, 8 | 13 | 18 | 23) != (byte == b'-'))
551        {
552            return Err(InvalidError::new(
553                "UUID must use the lowercase canonical 8-4-4-4-12 form",
554            ));
555        }
556        if value.bytes().any(|byte| byte.is_ascii_uppercase()) {
557            return Err(InvalidError::new(
558                "UUID hexadecimal digits must be lowercase",
559            ));
560        }
561
562        let mut bytes = [0u8; Self::BYTES];
563        let mut output = 0usize;
564        let mut nibbles = value.bytes().filter(|byte| *byte != b'-');
565        while output < bytes.len() {
566            let high = decode_hex(nibbles.next().expect("UUID shape checked"))?;
567            let low = decode_hex(nibbles.next().expect("UUID shape checked"))?;
568            bytes[output] = (high << 4) | low;
569            output += 1;
570        }
571        Ok(Self::new(bytes))
572    }
573}
574
575fn decode_hex(byte: u8) -> Result<u8, InvalidError> {
576    match byte {
577        b'0'..=b'9' => Ok(byte - b'0'),
578        b'a'..=b'f' => Ok(byte - b'a' + 10),
579        _ => Err(InvalidError::new("UUID contains a non-hexadecimal byte")),
580    }
581}
582
583#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
584pub struct BinaryValue(#[serde(with = "crate::encoding::bin_bytes")] pub Vec<u8>);
585
586#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
587#[serde(tag = "kind", content = "value", rename_all = "snake_case")]
588#[non_exhaustive]
589pub enum TypedValue {
590    Null,
591    Boolean(bool),
592    Int(i32),
593    Long(i64),
594    Float(f32),
595    Double(f64),
596    Decimal(DecimalValue),
597    Date(i32),
598    TimeMicros(i64),
599    TimestampMicros(i64),
600    TimestampTzMicros(i64),
601    String(String),
602    Uuid(UuidValue),
603    Fixed(BinaryValue),
604    Binary(BinaryValue),
605    Struct(Vec<FieldValue>),
606    List(Vec<TypedValue>),
607    Map(Vec<MapEntry>),
608}
609
610impl From<&str> for TypedValue {
611    fn from(value: &str) -> Self {
612        Self::String(value.to_owned())
613    }
614}
615
616impl From<String> for TypedValue {
617    fn from(value: String) -> Self {
618        Self::String(value)
619    }
620}
621
622impl From<&String> for TypedValue {
623    fn from(value: &String) -> Self {
624        Self::String(value.clone())
625    }
626}
627
628impl From<i32> for TypedValue {
629    fn from(value: i32) -> Self {
630        Self::Int(value)
631    }
632}
633
634impl From<u32> for TypedValue {
635    fn from(value: u32) -> Self {
636        Self::Long(i64::from(value))
637    }
638}
639
640impl From<i64> for TypedValue {
641    fn from(value: i64) -> Self {
642        Self::Long(value)
643    }
644}
645
646impl From<f32> for TypedValue {
647    fn from(value: f32) -> Self {
648        Self::Float(value)
649    }
650}
651
652impl From<f64> for TypedValue {
653    fn from(value: f64) -> Self {
654        Self::Double(value)
655    }
656}
657
658impl From<bool> for TypedValue {
659    fn from(value: bool) -> Self {
660        Self::Boolean(value)
661    }
662}
663
664impl<T: Into<TypedValue>> From<Vec<T>> for TypedValue {
665    fn from(values: Vec<T>) -> Self {
666        Self::List(values.into_iter().map(Into::into).collect())
667    }
668}
669
670impl TypedValue {
671    pub fn as_str(&self) -> Option<&str> {
672        match self {
673            Self::String(value) => Some(value),
674            _ => None,
675        }
676    }
677
678    pub fn as_u64(&self) -> Option<u64> {
679        match self {
680            Self::Int(value) => u64::try_from(*value).ok(),
681            Self::Long(value) => u64::try_from(*value).ok(),
682            _ => None,
683        }
684    }
685
686    pub fn as_i64(&self) -> Option<i64> {
687        match self {
688            Self::Int(value) => Some(i64::from(*value)),
689            Self::Long(value) => Some(*value),
690            _ => None,
691        }
692    }
693
694    pub fn diagnostic_text(&self) -> String {
695        match self {
696            Self::Null => "null".to_owned(),
697            Self::Boolean(value) => value.to_string(),
698            Self::Int(value) => value.to_string(),
699            Self::Long(value) => value.to_string(),
700            Self::Float(value) => value.to_string(),
701            Self::Double(value) => value.to_string(),
702            Self::Date(value) => value.to_string(),
703            Self::TimeMicros(value) => value.to_string(),
704            Self::TimestampMicros(value) => value.to_string(),
705            Self::TimestampTzMicros(value) => value.to_string(),
706            Self::String(value) => value.clone(),
707            Self::Uuid(value) => value.to_string(),
708            Self::Decimal(value) => {
709                format!("0x{} scale {}", encode_hex(&value.unscaled), value.scale)
710            }
711            Self::Fixed(value) | Self::Binary(value) => format!("0x{}", encode_hex(&value.0)),
712            Self::Struct(_) | Self::List(_) | Self::Map(_) => {
713                serde_json::to_string(self).unwrap_or_else(|_| "<invalid typed value>".to_owned())
714            }
715        }
716    }
717
718    pub fn validate_canonical(&self) -> Result<(), InvalidError> {
719        validate_typed_value(self, 1)
720    }
721
722    pub fn validate_against(
723        &self,
724        logical_type: &LogicalType,
725        required: bool,
726    ) -> Result<(), InvalidError> {
727        if matches!(self, Self::Null) {
728            return if required {
729                Err(InvalidError::new("required value is null"))
730            } else {
731                Ok(())
732            };
733        }
734
735        self.validate_canonical()?;
736
737        match (self, logical_type) {
738            (Self::Boolean(_), LogicalType::Boolean)
739            | (Self::Int(_), LogicalType::Int)
740            | (Self::Long(_), LogicalType::Long)
741            | (Self::Date(_), LogicalType::Date)
742            | (Self::TimestampMicros(_), LogicalType::TimestampMicros)
743            | (Self::TimestampTzMicros(_), LogicalType::TimestampTzMicros)
744            | (Self::String(_), LogicalType::String)
745            | (Self::Binary(_), LogicalType::Binary) => Ok(()),
746            (Self::Float(value), LogicalType::Float) => validate_float(*value as f64),
747            (Self::Double(value), LogicalType::Double) => validate_float(*value),
748            (Self::Decimal(value), LogicalType::Decimal { precision, scale }) => {
749                value.validate_canonical()?;
750                if value.precision != *precision || value.scale != *scale {
751                    return Err(InvalidError::new(format!(
752                        "decimal type mismatch: value is ({}, {}), schema is ({precision}, {scale})",
753                        value.precision, value.scale
754                    )));
755                }
756                Ok(())
757            }
758            (Self::TimeMicros(value), LogicalType::TimeMicros) => {
759                if !(0..MICROS_PER_DAY).contains(value) {
760                    return Err(InvalidError::new(format!(
761                        "time value {value} is outside one day in microseconds"
762                    )));
763                }
764                Ok(())
765            }
766            (Self::Uuid(value), LogicalType::Uuid) => {
767                if value.0.len() != UuidValue::BYTES {
768                    return Err(InvalidError::new(format!(
769                        "UUID must be {} bytes, got {}",
770                        UuidValue::BYTES,
771                        value.0.len()
772                    )));
773                }
774                Ok(())
775            }
776            (Self::Fixed(value), LogicalType::Fixed { length }) => {
777                if value.0.len() != *length as usize {
778                    return Err(InvalidError::new(format!(
779                        "fixed value is {} bytes, expected {length}",
780                        value.0.len()
781                    )));
782                }
783                Ok(())
784            }
785            (Self::Struct(values), LogicalType::Struct { fields }) => {
786                validate_struct_value(values, fields)
787            }
788            (
789                Self::List(values),
790                LogicalType::List {
791                    element_required,
792                    element,
793                    ..
794                },
795            ) => {
796                for value in values {
797                    value.validate_against(element, *element_required)?;
798                }
799                Ok(())
800            }
801            (
802                Self::Map(entries),
803                LogicalType::Map {
804                    key,
805                    value,
806                    value_required,
807                    ..
808                },
809            ) => validate_map(entries, key, value, *value_required),
810            (_, LogicalType::Variant) => Ok(()),
811            _ => Err(InvalidError::new("typed value does not match logical type")),
812        }
813    }
814}
815
816fn encode_hex(bytes: &[u8]) -> String {
817    let mut encoded = String::with_capacity(bytes.len() * 2);
818    for byte in bytes {
819        use std::fmt::Write as _;
820        let _ = write!(encoded, "{byte:02x}");
821    }
822    encoded
823}
824
825#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
826pub struct FieldValue {
827    pub field_id: u32,
828    pub value: TypedValue,
829}
830
831#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
832pub struct MapEntry {
833    pub key: TypedValue,
834    pub value: TypedValue,
835}
836
837impl Validate for TypedValue {
838    fn validate(&self) -> Result<(), InvalidError> {
839        self.validate_canonical()
840    }
841}
842
843fn validate_typed_value(value: &TypedValue, depth: usize) -> Result<(), InvalidError> {
844    validate_depth(depth)?;
845    match value {
846        TypedValue::Null
847        | TypedValue::Boolean(_)
848        | TypedValue::Int(_)
849        | TypedValue::Long(_)
850        | TypedValue::Date(_)
851        | TypedValue::TimestampMicros(_)
852        | TypedValue::TimestampTzMicros(_) => Ok(()),
853        TypedValue::String(value) => {
854            if value.len() > MAX_VALUE_BYTES {
855                return Err(InvalidError::new(format!(
856                    "string value is {} bytes, exceeds cap {MAX_VALUE_BYTES}",
857                    value.len()
858                )));
859            }
860            Ok(())
861        }
862        TypedValue::Float(value) => validate_float(f64::from(*value)),
863        TypedValue::Double(value) => validate_float(*value),
864        TypedValue::Decimal(value) => value.validate_canonical(),
865        TypedValue::TimeMicros(value) => {
866            if !(0..MICROS_PER_DAY).contains(value) {
867                return Err(InvalidError::new(format!(
868                    "time value {value} is outside one day in microseconds"
869                )));
870            }
871            Ok(())
872        }
873        TypedValue::Uuid(value) => {
874            if value.0.len() != UuidValue::BYTES {
875                return Err(InvalidError::new(format!(
876                    "UUID must be {} bytes, got {}",
877                    UuidValue::BYTES,
878                    value.0.len()
879                )));
880            }
881            Ok(())
882        }
883        TypedValue::Fixed(value) => {
884            if value.0.len() > MAX_FIXED_BYTES as usize {
885                return Err(InvalidError::new(format!(
886                    "binary value is {} bytes, exceeds cap {MAX_FIXED_BYTES}",
887                    value.0.len()
888                )));
889            }
890            Ok(())
891        }
892        TypedValue::Binary(value) => {
893            if value.0.len() > MAX_VALUE_BYTES {
894                return Err(InvalidError::new(format!(
895                    "binary value is {} bytes, exceeds cap {MAX_VALUE_BYTES}",
896                    value.0.len()
897                )));
898            }
899            Ok(())
900        }
901        TypedValue::Struct(values) => {
902            if values.len() > MAX_LOGICAL_SCHEMA_FIELDS {
903                return Err(InvalidError::new(format!(
904                    "struct carries {} fields, exceeds cap {MAX_LOGICAL_SCHEMA_FIELDS}",
905                    values.len()
906                )));
907            }
908            let mut previous_id = 0;
909            for value in values {
910                if value.field_id == 0 || value.field_id <= previous_id {
911                    return Err(InvalidError::new(
912                        "struct field values must have strictly increasing positive field IDs",
913                    ));
914                }
915                previous_id = value.field_id;
916                validate_typed_value(&value.value, depth + 1)?;
917            }
918            Ok(())
919        }
920        TypedValue::List(values) => {
921            if values.len() > MAX_LOGICAL_SCHEMA_FIELDS {
922                return Err(InvalidError::new(format!(
923                    "list carries {} values, exceeds cap {MAX_LOGICAL_SCHEMA_FIELDS}",
924                    values.len()
925                )));
926            }
927            for value in values {
928                validate_typed_value(value, depth + 1)?;
929            }
930            Ok(())
931        }
932        TypedValue::Map(entries) => {
933            if entries.len() > MAX_LOGICAL_SCHEMA_FIELDS {
934                return Err(InvalidError::new(format!(
935                    "map carries {} entries, exceeds cap {MAX_LOGICAL_SCHEMA_FIELDS}",
936                    entries.len()
937                )));
938            }
939            let mut previous = None;
940            for entry in entries {
941                validate_typed_value(&entry.key, depth + 1)?;
942                validate_typed_value(&entry.value, depth + 1)?;
943                let key = canonical_map_key(&entry.key)?;
944                if previous.as_ref().is_some_and(|before| before >= &key) {
945                    return Err(InvalidError::new(
946                        "map entries must be strictly ordered by canonical key",
947                    ));
948                }
949                previous = Some(key);
950            }
951            Ok(())
952        }
953    }
954}
955
956fn validate_struct_fields(
957    fields: &[LogicalField],
958    depth: usize,
959    ids: &mut BTreeSet<u32>,
960    field_count: &mut usize,
961    encoded_size: &mut usize,
962    allow_provenance: bool,
963) -> Result<(), InvalidError> {
964    if depth > MAX_LOGICAL_SCHEMA_DEPTH {
965        return Err(InvalidError::new(format!(
966            "logical schema depth {depth} exceeds cap {MAX_LOGICAL_SCHEMA_DEPTH}"
967        )));
968    }
969
970    let mut names = BTreeSet::new();
971    for field in fields {
972        validate_field_identity(field.id, &field.name, ids, &mut names, allow_provenance)?;
973        *field_count += 1;
974        if *field_count > MAX_LOGICAL_SCHEMA_FIELDS {
975            return Err(InvalidError::new(format!(
976                "logical schema field count exceeds cap {MAX_LOGICAL_SCHEMA_FIELDS}"
977            )));
978        }
979
980        let doc_bytes = field.doc.as_ref().map_or(0, String::len);
981        if doc_bytes > MAX_FIELD_DOC_BYTES {
982            return Err(InvalidError::new(format!(
983                "field `{}` documentation is {doc_bytes} bytes, exceeds cap {MAX_FIELD_DOC_BYTES}",
984                field.name
985            )));
986        }
987        *encoded_size = encoded_size
988            .checked_add(field.name.len() + doc_bytes + 16)
989            .ok_or_else(|| InvalidError::new("logical schema size overflow"))?;
990        if *encoded_size > MAX_LOGICAL_SCHEMA_BYTES {
991            return Err(InvalidError::new(format!(
992                "logical schema exceeds cap {MAX_LOGICAL_SCHEMA_BYTES} bytes"
993            )));
994        }
995        validate_logical_type(&field.field_type, depth, ids, field_count, encoded_size)?;
996    }
997    Ok(())
998}
999
1000fn validate_field_identity(
1001    id: u32,
1002    name: &str,
1003    ids: &mut BTreeSet<u32>,
1004    names: &mut BTreeSet<String>,
1005    allow_provenance: bool,
1006) -> Result<(), InvalidError> {
1007    let provenance_pair = PROVENANCE_FIELDS
1008        .iter()
1009        .any(|(reserved_id, reserved_name)| *reserved_id == id && *reserved_name == name);
1010    if id == 0 || (id >= PROVENANCE_FIELD_ID_START && !(allow_provenance && provenance_pair)) {
1011        return Err(InvalidError::new(format!(
1012            "field id {id} is not in the user field range"
1013        )));
1014    }
1015    if !ids.insert(id) {
1016        return Err(InvalidError::new(format!("duplicate field id {id}")));
1017    }
1018    validate_field_name(name)?;
1019    if PROVENANCE_FIELDS
1020        .iter()
1021        .any(|(_, reserved)| *reserved == name)
1022        && !(allow_provenance && provenance_pair)
1023    {
1024        return Err(InvalidError::new(format!(
1025            "field name `{name}` is reserved for provenance"
1026        )));
1027    }
1028    if !names.insert(name.to_owned()) {
1029        return Err(InvalidError::new(format!(
1030            "duplicate field name `{name}` in one struct"
1031        )));
1032    }
1033    Ok(())
1034}
1035
1036fn validate_field_name(name: &str) -> Result<(), InvalidError> {
1037    if name.is_empty() {
1038        return Err(InvalidError::new("field name must not be empty"));
1039    }
1040    if name.len() > MAX_FIELD_NAME_BYTES {
1041        return Err(InvalidError::new(format!(
1042            "field name is {} bytes, exceeds cap {MAX_FIELD_NAME_BYTES}",
1043            name.len()
1044        )));
1045    }
1046    if name.trim() != name || name.chars().any(char::is_control) {
1047        return Err(InvalidError::new(format!(
1048            "field name `{name}` contains surrounding whitespace or control characters"
1049        )));
1050    }
1051    Ok(())
1052}
1053
1054fn validate_logical_type(
1055    logical_type: &LogicalType,
1056    depth: usize,
1057    ids: &mut BTreeSet<u32>,
1058    field_count: &mut usize,
1059    encoded_size: &mut usize,
1060) -> Result<(), InvalidError> {
1061    match logical_type {
1062        LogicalType::Decimal { precision, scale } => validate_precision_scale(*precision, *scale),
1063        LogicalType::Fixed { length } => {
1064            if *length == 0 || *length > MAX_FIXED_BYTES {
1065                return Err(InvalidError::new(format!(
1066                    "fixed length {length} is outside 1..={MAX_FIXED_BYTES}"
1067                )));
1068            }
1069            Ok(())
1070        }
1071        LogicalType::Struct { fields } => {
1072            validate_struct_fields(fields, depth + 1, ids, field_count, encoded_size, false)
1073        }
1074        LogicalType::List {
1075            element_id,
1076            element,
1077            ..
1078        } => {
1079            validate_nested_id(*element_id, "list element", ids, field_count)?;
1080            validate_depth(depth + 1)?;
1081            validate_logical_type(element, depth + 1, ids, field_count, encoded_size)
1082        }
1083        LogicalType::Map {
1084            key_id,
1085            key,
1086            value_id,
1087            value,
1088            ..
1089        } => {
1090            validate_nested_id(*key_id, "map key", ids, field_count)?;
1091            validate_nested_id(*value_id, "map value", ids, field_count)?;
1092            if !key.accepts_map_key() {
1093                return Err(InvalidError::new(
1094                    "map key type must be a deterministic non-floating primitive",
1095                ));
1096            }
1097            validate_depth(depth + 1)?;
1098            validate_logical_type(key, depth + 1, ids, field_count, encoded_size)?;
1099            validate_logical_type(value, depth + 1, ids, field_count, encoded_size)
1100        }
1101        _ => Ok(()),
1102    }
1103}
1104
1105fn validate_nested_id(
1106    id: u32,
1107    label: &str,
1108    ids: &mut BTreeSet<u32>,
1109    field_count: &mut usize,
1110) -> Result<(), InvalidError> {
1111    if id == 0 || id >= PROVENANCE_FIELD_ID_START {
1112        return Err(InvalidError::new(format!(
1113            "{label} id {id} is not in the user field range"
1114        )));
1115    }
1116    if !ids.insert(id) {
1117        return Err(InvalidError::new(format!("duplicate {label} id {id}")));
1118    }
1119    *field_count += 1;
1120    if *field_count > MAX_LOGICAL_SCHEMA_FIELDS {
1121        return Err(InvalidError::new(format!(
1122            "logical schema field count exceeds cap {MAX_LOGICAL_SCHEMA_FIELDS}"
1123        )));
1124    }
1125    Ok(())
1126}
1127
1128fn validate_depth(depth: usize) -> Result<(), InvalidError> {
1129    if depth > MAX_LOGICAL_SCHEMA_DEPTH {
1130        return Err(InvalidError::new(format!(
1131            "logical schema depth {depth} exceeds cap {MAX_LOGICAL_SCHEMA_DEPTH}"
1132        )));
1133    }
1134    Ok(())
1135}
1136
1137fn validate_precision_scale(precision: u8, scale: u8) -> Result<(), InvalidError> {
1138    if precision == 0 || precision > MAX_DECIMAL_PRECISION || scale > precision {
1139        return Err(InvalidError::new(format!(
1140            "decimal precision and scale ({precision}, {scale}) are invalid"
1141        )));
1142    }
1143    Ok(())
1144}
1145
1146fn validate_decimal(value: &DecimalValue) -> Result<(), InvalidError> {
1147    validate_precision_scale(value.precision, value.scale)?;
1148    if value.unscaled.is_empty() || value.unscaled.len() > 16 {
1149        return Err(InvalidError::new(format!(
1150            "decimal unscaled value must contain 1..=16 bytes, got {}",
1151            value.unscaled.len()
1152        )));
1153    }
1154    if value.unscaled.len() > 1 {
1155        let first = value.unscaled[0];
1156        let second = value.unscaled[1];
1157        if (first == 0 && second & 0x80 == 0) || (first == 0xff && second & 0x80 != 0) {
1158            return Err(InvalidError::new(
1159                "decimal unscaled bytes are not minimal two's complement",
1160            ));
1161        }
1162    }
1163
1164    let fill = if value.unscaled[0] & 0x80 == 0 {
1165        0
1166    } else {
1167        0xff
1168    };
1169    let mut bytes = [fill; 16];
1170    let start = bytes.len() - value.unscaled.len();
1171    bytes[start..].copy_from_slice(&value.unscaled);
1172    let unscaled = i128::from_be_bytes(bytes);
1173    let digits = if unscaled == 0 {
1174        1
1175    } else {
1176        unscaled.unsigned_abs().ilog10() + 1
1177    };
1178    if digits > u32::from(value.precision) {
1179        return Err(InvalidError::new(format!(
1180            "decimal value has {digits} digits, exceeds precision {}",
1181            value.precision
1182        )));
1183    }
1184    Ok(())
1185}
1186
1187fn validate_float(value: f64) -> Result<(), InvalidError> {
1188    if !value.is_finite() {
1189        return Err(InvalidError::new("floating-point value must be finite"));
1190    }
1191    if value == 0.0 && value.is_sign_negative() {
1192        return Err(InvalidError::new(
1193            "negative zero is not a canonical floating-point value",
1194        ));
1195    }
1196    Ok(())
1197}
1198
1199fn validate_struct_value(
1200    values: &[FieldValue],
1201    fields: &[LogicalField],
1202) -> Result<(), InvalidError> {
1203    if values.len() != fields.len() {
1204        return Err(InvalidError::new(format!(
1205            "struct carries {} values, schema has {} fields",
1206            values.len(),
1207            fields.len()
1208        )));
1209    }
1210    for (value, field) in values.iter().zip(fields) {
1211        if value.field_id != field.id {
1212            return Err(InvalidError::new(format!(
1213                "struct field id {} appears where {} is required",
1214                value.field_id, field.id
1215            )));
1216        }
1217        value
1218            .value
1219            .validate_against(&field.field_type, field.required)?;
1220    }
1221    Ok(())
1222}
1223
1224fn validate_map(
1225    entries: &[MapEntry],
1226    key_type: &LogicalType,
1227    value_type: &LogicalType,
1228    value_required: bool,
1229) -> Result<(), InvalidError> {
1230    let mut previous = None;
1231    for entry in entries {
1232        entry.key.validate_against(key_type, true)?;
1233        entry.value.validate_against(value_type, value_required)?;
1234        let key = canonical_map_key(&entry.key)?;
1235        if previous.as_ref().is_some_and(|before| before >= &key) {
1236            return Err(InvalidError::new(
1237                "map entries must be strictly ordered by canonical key",
1238            ));
1239        }
1240        previous = Some(key);
1241    }
1242    Ok(())
1243}
1244
1245fn canonical_map_key(value: &TypedValue) -> Result<Vec<u8>, InvalidError> {
1246    let mut key = Vec::new();
1247    match value {
1248        TypedValue::Boolean(value) => {
1249            key.push(0);
1250            key.push(u8::from(*value));
1251        }
1252        TypedValue::Int(value) => {
1253            key.push(1);
1254            key.extend_from_slice(&value.to_be_bytes());
1255        }
1256        TypedValue::Long(value) => {
1257            key.push(2);
1258            key.extend_from_slice(&value.to_be_bytes());
1259        }
1260        TypedValue::Decimal(value) => {
1261            key.push(3);
1262            key.push(value.precision);
1263            key.push(value.scale);
1264            key.extend_from_slice(&value.unscaled);
1265        }
1266        TypedValue::Date(value) => {
1267            key.push(4);
1268            key.extend_from_slice(&value.to_be_bytes());
1269        }
1270        TypedValue::TimeMicros(value) => {
1271            key.push(5);
1272            key.extend_from_slice(&value.to_be_bytes());
1273        }
1274        TypedValue::TimestampMicros(value) => {
1275            key.push(6);
1276            key.extend_from_slice(&value.to_be_bytes());
1277        }
1278        TypedValue::TimestampTzMicros(value) => {
1279            key.push(7);
1280            key.extend_from_slice(&value.to_be_bytes());
1281        }
1282        TypedValue::String(value) => {
1283            key.push(8);
1284            key.extend_from_slice(value.as_bytes());
1285        }
1286        TypedValue::Uuid(value) => {
1287            key.push(9);
1288            key.extend_from_slice(&value.0);
1289        }
1290        TypedValue::Fixed(value) => {
1291            key.push(10);
1292            key.extend_from_slice(&value.0);
1293        }
1294        TypedValue::Binary(value) => {
1295            key.push(11);
1296            key.extend_from_slice(&value.0);
1297        }
1298        _ => {
1299            return Err(InvalidError::new(
1300                "map key must be a deterministic non-floating primitive",
1301            ));
1302        }
1303    }
1304    Ok(key)
1305}
1306
1307#[cfg(test)]
1308mod tests {
1309    use super::*;
1310
1311    fn fields() -> Vec<LogicalField> {
1312        vec![
1313            LogicalField {
1314                id: 1,
1315                name: "id".to_owned(),
1316                required: true,
1317                field_type: LogicalType::Uuid,
1318                doc: Some("Stable row identity".to_owned()),
1319            },
1320            LogicalField {
1321                id: 2,
1322                name: "attributes".to_owned(),
1323                required: false,
1324                field_type: LogicalType::Map {
1325                    key_id: 3,
1326                    key: Box::new(LogicalType::String),
1327                    value_id: 4,
1328                    value_required: true,
1329                    value: Box::new(LogicalType::Long),
1330                },
1331                doc: None,
1332            },
1333        ]
1334    }
1335
1336    #[test]
1337    fn given_a_logical_schema_when_fingerprinted_then_should_be_deterministic_and_cover_metadata() {
1338        let schema = LogicalSchema::new(LogicalSchemaId::from_u128(1), 1, fields())
1339            .expect("schema is valid");
1340        assert_eq!(
1341            schema.schema.fingerprint,
1342            schema.compute_fingerprint().expect("fingerprint")
1343        );
1344        schema.validate().expect("fingerprint validates");
1345
1346        let mut changed_fields = fields();
1347        changed_fields[0].doc = Some("Different documentation".to_owned());
1348        let changed = LogicalSchema::new(LogicalSchemaId::from_u128(1), 1, changed_fields)
1349            .expect("changed schema is valid");
1350        assert_ne!(schema.schema.fingerprint, changed.schema.fingerprint);
1351    }
1352
1353    #[test]
1354    fn given_reserved_or_duplicate_nested_ids_when_a_schema_is_validated_then_should_reject() {
1355        let reserved = vec![LogicalField {
1356            id: PROVENANCE_FIELDS[0].0,
1357            name: PROVENANCE_FIELDS[0].1.to_owned(),
1358            required: true,
1359            field_type: LogicalType::String,
1360            doc: None,
1361        }];
1362        assert!(LogicalSchema::new(LogicalSchemaId::from_u128(1), 1, reserved).is_err());
1363
1364        let duplicate = vec![LogicalField {
1365            id: 1,
1366            name: "items".to_owned(),
1367            required: true,
1368            field_type: LogicalType::List {
1369                element_id: 1,
1370                element_required: true,
1371                element: Box::new(LogicalType::String),
1372            },
1373            doc: None,
1374        }];
1375        assert!(LogicalSchema::new(LogicalSchemaId::from_u128(1), 1, duplicate).is_err());
1376    }
1377
1378    #[test]
1379    fn given_typed_values_when_validated_then_should_enforce_canonical_decimal_float_time_and_map_forms()
1380     {
1381        assert!(TypedValue::Double(f64::NAN).validate().is_err());
1382        assert!(TypedValue::Float(-0.0).validate().is_err());
1383        assert!(TypedValue::TimeMicros(MICROS_PER_DAY).validate().is_err());
1384        assert!(
1385            TypedValue::Decimal(DecimalValue {
1386                unscaled: vec![0, 1],
1387                precision: 2,
1388                scale: 0,
1389            })
1390            .validate()
1391            .is_err()
1392        );
1393        assert!(
1394            TypedValue::Map(vec![
1395                MapEntry {
1396                    key: TypedValue::String("b".to_owned()),
1397                    value: TypedValue::Long(1),
1398                },
1399                MapEntry {
1400                    key: TypedValue::String("a".to_owned()),
1401                    value: TypedValue::Long(2),
1402                },
1403            ])
1404            .validate()
1405            .is_err()
1406        );
1407    }
1408
1409    #[test]
1410    fn given_a_variant_field_when_validated_then_should_accept_every_canonical_value_and_refuse_a_map_key()
1411     {
1412        for value in [
1413            TypedValue::Null,
1414            TypedValue::Boolean(true),
1415            TypedValue::Long(i64::MAX),
1416            TypedValue::Double(2.5),
1417            TypedValue::String("text".to_owned()),
1418        ] {
1419            value
1420                .validate_against(&LogicalType::Variant, false)
1421                .expect("a variant takes any tagged value");
1422        }
1423        assert!(
1424            TypedValue::Double(f64::NAN)
1425                .validate_against(&LogicalType::Variant, false)
1426                .is_err()
1427        );
1428        assert!(
1429            TypedValue::Null
1430                .validate_against(&LogicalType::Variant, true)
1431                .is_err()
1432        );
1433        assert!(!LogicalType::Variant.accepts_map_key());
1434        assert_eq!(LogicalType::Variant.kind(), LogicalTypeKind::Variant);
1435        let json = serde_json::to_string(&LogicalType::Variant).expect("type serializes");
1436        assert_eq!(json, r#"{"kind":"variant"}"#);
1437        let mut variant_fields = fields();
1438        variant_fields[1].field_type = LogicalType::Variant;
1439        let variant = LogicalSchema::new(LogicalSchemaId::from_u128(1), 1, variant_fields)
1440            .expect("a variant schema is valid");
1441        let map = LogicalSchema::new(LogicalSchemaId::from_u128(1), 1, fields())
1442            .expect("the map schema is valid");
1443        assert_ne!(variant.schema.fingerprint, map.schema.fingerprint);
1444    }
1445
1446    #[test]
1447    fn given_ids_and_binary_values_when_round_tripped_through_json_then_should_be_lossless() {
1448        let id = LogicalSchemaId::from_u128(u128::MAX - 7);
1449        let json = serde_json::to_string(&id).expect("id serializes");
1450        let decoded: LogicalSchemaId = serde_json::from_str(&json).expect("id deserializes");
1451        assert_eq!(decoded, id);
1452
1453        let value = TypedValue::Binary(BinaryValue(vec![0, 127, 255]));
1454        let json = serde_json::to_string(&value).expect("value serializes");
1455        let decoded: TypedValue = serde_json::from_str(&json).expect("value deserializes");
1456        assert_eq!(decoded, value);
1457    }
1458}
laser_wire/schema.rs

laser_wire/
schema.rs
