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

laser_wire/
schema.rs
