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