cobble_table/
codec.rs

1use crate::{LogicalType, LogicalTypeKind, Result, TableError, TimestampKind};
2use bytes::Bytes;
3use std::ops::Range;
4
5const NANOS_PER_DAY: i64 = 86_400_000_000_000;
6
7/// Schema-directed dynamic value used by the `cobble-table-v1` codec.
8#[derive(Clone, Debug, PartialEq)]
9pub enum Value {
10    Null,
11    Boolean(bool),
12    Int8(i8),
13    Int16(i16),
14    Int32(i32),
15    Int64(i64),
16    Float32(f32),
17    Float64(f64),
18    Decimal {
19        precision: u8,
20        scale: u8,
21        unscaled: i128,
22    },
23    Date(i32),
24    Time(i64),
25    Timestamp {
26        precision: u8,
27        timestamp_kind: TimestampKind,
28        seconds: i64,
29        nanos: u32,
30    },
31    String(String),
32    Binary(Bytes),
33    List(Vec<Value>),
34    Map(Vec<(Value, Value)>),
35    Struct(Vec<Value>),
36    Extension {
37        type_id: String,
38        value: Box<Value>,
39    },
40}
41
42impl From<Vec<u8>> for Value {
43    fn from(value: Vec<u8>) -> Self {
44        Self::Binary(Bytes::from(value))
45    }
46}
47
48impl From<Bytes> for Value {
49    fn from(value: Bytes) -> Self {
50        Self::Binary(value)
51    }
52}
53
54/// Ordered, schema-directed composite-key codec.
55pub struct KeyCodec;
56
57impl KeyCodec {
58    pub fn encode_row(types: &[LogicalType], values: &[Value]) -> Result<Vec<u8>> {
59        let mut out = Vec::new();
60        Self::encode_row_into(types, values, &mut out)?;
61        Ok(out)
62    }
63
64    pub fn encode_row_into(
65        types: &[LogicalType],
66        values: &[Value],
67        out: &mut Vec<u8>,
68    ) -> Result<()> {
69        let start = out.len();
70        let result = Self::append_row(types, values, out);
71        if result.is_err() {
72            out.truncate(start);
73        }
74        result
75    }
76
77    /// Encodes a row whose logical types were already validated when its table was opened.
78    pub(crate) fn encode_row_with_prefix_validated(
79        types: &[LogicalType],
80        values: &[Value],
81        prefix_fields: usize,
82        out: &mut Vec<u8>,
83    ) -> Result<usize> {
84        let start = out.len();
85        let result = (|| {
86            if types.len() != values.len() {
87                return Err(TableError::codec(
88                    "key field count does not match value count",
89                ));
90            }
91            debug_assert!(prefix_fields <= types.len());
92            let mut prefix_end = start;
93            for (index, (logical_type, value)) in types.iter().zip(values).enumerate() {
94                encode_key(logical_type, value, out)?;
95                if index + 1 == prefix_fields {
96                    prefix_end = out.len();
97                }
98            }
99            Ok(prefix_end - start)
100        })();
101        if result.is_err() {
102            out.truncate(start);
103        }
104        result
105    }
106
107    /// Encodes selected fields from a row whose logical types were validated at table creation.
108    pub(crate) fn encode_row_from_positions_validated(
109        types: &[LogicalType],
110        row: &[Value],
111        positions: &[usize],
112        prefix_fields: usize,
113    ) -> Result<(Vec<u8>, usize)> {
114        debug_assert_eq!(types.len(), positions.len());
115        debug_assert!(prefix_fields <= types.len());
116        let mut encoded = Vec::new();
117        let mut prefix_end = 0;
118        for (index, (logical_type, position)) in types.iter().zip(positions).enumerate() {
119            encode_key(logical_type, &row[*position], &mut encoded)?;
120            if index + 1 == prefix_fields {
121                prefix_end = encoded.len();
122            }
123        }
124        Ok((encoded, prefix_end))
125    }
126
127    fn append_row(types: &[LogicalType], values: &[Value], out: &mut Vec<u8>) -> Result<()> {
128        if types.len() != values.len() {
129            return Err(TableError::codec(
130                "key field count does not match value count",
131            ));
132        }
133        for (logical_type, value) in types.iter().zip(values) {
134            logical_type.validate()?;
135            encode_key(logical_type, value, out)?;
136        }
137        Ok(())
138    }
139
140    pub fn encode_scalar(logical_type: &LogicalType, value: &Value) -> Result<Vec<u8>> {
141        let mut out = Vec::new();
142        logical_type.validate()?;
143        encode_key(logical_type, value, &mut out)?;
144        Ok(out)
145    }
146
147    pub fn decode_row(types: &[LogicalType], encoded: &[u8]) -> Result<Vec<Value>> {
148        for logical_type in types {
149            logical_type.validate()?;
150        }
151        Self::decode_row_validated(types, encoded)
152    }
153
154    /// Decodes a key with types validated at table creation.
155    pub(crate) fn decode_row_validated(
156        types: &[LogicalType],
157        encoded: &[u8],
158    ) -> Result<Vec<Value>> {
159        let mut offset = 0;
160        let mut values = Vec::with_capacity(types.len());
161        for logical_type in types {
162            let (value, consumed) = decode_key(logical_type, &encoded[offset..])?;
163            offset += consumed;
164            values.push(value);
165        }
166        ensure_consumed(encoded.len(), offset)?;
167        Ok(values)
168    }
169
170    /// Decodes a key directly into its row positions for an already validated table layout.
171    pub(crate) fn decode_row_into_positions_validated(
172        types: &[LogicalType],
173        encoded: &[u8],
174        positions: &[usize],
175        row: &mut [Value],
176    ) -> Result<()> {
177        debug_assert_eq!(types.len(), positions.len());
178        debug_assert!(positions.iter().all(|position| *position < row.len()));
179        let mut offset = 0;
180        for (logical_type, position) in types.iter().zip(positions) {
181            let (value, consumed) = decode_key(logical_type, &encoded[offset..])?;
182            row[*position] = value;
183            offset += consumed;
184        }
185        ensure_consumed(encoded.len(), offset)
186    }
187
188    pub fn decode_scalar(logical_type: &LogicalType, encoded: &[u8]) -> Result<Value> {
189        logical_type.validate()?;
190        let (value, consumed) = decode_key(logical_type, encoded)?;
191        ensure_consumed(encoded.len(), consumed)?;
192        Ok(value)
193    }
194}
195
196/// Schema-directed value codec. `encode_into` appends to caller-owned memory.
197pub struct ValueCodec;
198
199impl ValueCodec {
200    pub fn encode(logical_type: &LogicalType, value: &Value) -> Result<Vec<u8>> {
201        let mut out = Vec::new();
202        Self::encode_into(logical_type, value, &mut out)?;
203        Ok(out)
204    }
205
206    pub fn encode_into(logical_type: &LogicalType, value: &Value, out: &mut Vec<u8>) -> Result<()> {
207        let start = out.len();
208        let result = logical_type
209            .validate()
210            .and_then(|_| encode_value(logical_type, value, out));
211        if result.is_err() {
212            out.truncate(start);
213        }
214        result
215    }
216
217    /// Encodes a value whose logical type was validated at table creation.
218    pub(crate) fn encode_validated(logical_type: &LogicalType, value: &Value) -> Result<Vec<u8>> {
219        let mut out = Vec::new();
220        encode_value(logical_type, value, &mut out)?;
221        Ok(out)
222    }
223
224    pub fn encoded_size(logical_type: &LogicalType, value: &Value) -> Result<usize> {
225        logical_type.validate()?;
226        value_size(logical_type, value)
227    }
228
229    pub fn decode(logical_type: &LogicalType, encoded: &[u8]) -> Result<Value> {
230        logical_type.validate()?;
231        decode_value(logical_type, Input::Borrowed(encoded))
232    }
233
234    /// Decode an owned buffer. Every `Binary` value, including nested values, keeps a zero-copy
235    /// slice of `encoded`; retain the returned value only while retaining that allocation.
236    pub fn decode_bytes(logical_type: &LogicalType, encoded: Bytes) -> Result<Value> {
237        logical_type.validate()?;
238        decode_value(logical_type, Input::Owned(encoded))
239    }
240
241    /// Decodes a value with a logical type validated at table creation.
242    pub(crate) fn decode_bytes_validated(
243        logical_type: &LogicalType,
244        encoded: Bytes,
245    ) -> Result<Value> {
246        decode_value(logical_type, Input::Owned(encoded))
247    }
248}
249
250enum Input<'a> {
251    Borrowed(&'a [u8]),
252    Owned(Bytes),
253}
254
255impl Input<'_> {
256    fn bytes(&self) -> &[u8] {
257        match self {
258            Self::Borrowed(bytes) => bytes,
259            Self::Owned(bytes) => bytes,
260        }
261    }
262
263    fn slice(&self, range: Range<usize>) -> Self {
264        match self {
265            Self::Borrowed(bytes) => Self::Borrowed(&bytes[range]),
266            Self::Owned(bytes) => Self::Owned(bytes.slice(range)),
267        }
268    }
269
270    fn into_bytes(self) -> Bytes {
271        match self {
272            Self::Borrowed(bytes) => Bytes::copy_from_slice(bytes),
273            Self::Owned(bytes) => bytes,
274        }
275    }
276}
277
278fn encode_key(logical_type: &LogicalType, value: &Value, out: &mut Vec<u8>) -> Result<()> {
279    if !logical_type.is_key_compatible() {
280        return Err(TableError::codec("type cannot be used as a key"));
281    }
282    match (&logical_type.kind, value) {
283        (LogicalTypeKind::Boolean, Value::Boolean(value)) => out.push(u8::from(*value)),
284        (LogicalTypeKind::Int8, Value::Int8(value)) => ordered_i128(*value as i128, 1, out),
285        (LogicalTypeKind::Int16, Value::Int16(value)) => ordered_i128(*value as i128, 2, out),
286        (LogicalTypeKind::Int32, Value::Int32(value)) => ordered_i128(*value as i128, 4, out),
287        (LogicalTypeKind::Int64, Value::Int64(value)) => ordered_i128(*value as i128, 8, out),
288        (
289            LogicalTypeKind::Decimal { precision, scale },
290            Value::Decimal {
291                precision: actual_precision,
292                scale: actual_scale,
293                unscaled,
294            },
295        ) => {
296            validate_decimal(
297                *precision,
298                *scale,
299                *actual_precision,
300                *actual_scale,
301                *unscaled,
302            )?;
303            ordered_i128(*unscaled, decimal_width(*precision), out);
304        }
305        (LogicalTypeKind::Date, Value::Date(value)) => ordered_i128(*value as i128, 4, out),
306        (LogicalTypeKind::Time { precision }, Value::Time(value)) => {
307            validate_time(*value, *precision)?;
308            ordered_i128(*value as i128, 8, out);
309        }
310        (
311            LogicalTypeKind::Timestamp {
312                precision,
313                timestamp_kind,
314            },
315            Value::Timestamp {
316                precision: actual_precision,
317                timestamp_kind: actual_kind,
318                seconds,
319                nanos,
320            },
321        ) => {
322            validate_timestamp(
323                *precision,
324                *timestamp_kind,
325                *actual_precision,
326                *actual_kind,
327                *seconds,
328                *nanos,
329            )?;
330            ordered_i128(*seconds as i128, 8, out);
331            out.extend_from_slice(&nanos.to_be_bytes());
332        }
333        (LogicalTypeKind::String, Value::String(value)) => append_escaped(value.as_bytes(), out),
334        (LogicalTypeKind::Binary, Value::Binary(value)) => append_escaped(value, out),
335        _ => return Err(type_mismatch(logical_type, value)),
336    }
337    Ok(())
338}
339
340fn decode_key(logical_type: &LogicalType, encoded: &[u8]) -> Result<(Value, usize)> {
341    if !logical_type.is_key_compatible() {
342        return Err(TableError::codec("type cannot be used as a key"));
343    }
344    match &logical_type.kind {
345        LogicalTypeKind::Boolean => Ok((Value::Boolean(read_bool(encoded)?), 1)),
346        LogicalTypeKind::Int8 => Ok((Value::Int8(read_ordered(encoded, 1)? as i8), 1)),
347        LogicalTypeKind::Int16 => Ok((Value::Int16(read_ordered(encoded, 2)? as i16), 2)),
348        LogicalTypeKind::Int32 => Ok((Value::Int32(read_ordered(encoded, 4)? as i32), 4)),
349        LogicalTypeKind::Int64 => Ok((Value::Int64(read_ordered(encoded, 8)? as i64), 8)),
350        LogicalTypeKind::Decimal { precision, scale } => {
351            let width = decimal_width(*precision);
352            let unscaled = read_ordered(encoded, width)?;
353            validate_decimal(*precision, *scale, *precision, *scale, unscaled)?;
354            Ok((
355                Value::Decimal {
356                    precision: *precision,
357                    scale: *scale,
358                    unscaled,
359                },
360                width,
361            ))
362        }
363        LogicalTypeKind::Date => Ok((Value::Date(read_ordered(encoded, 4)? as i32), 4)),
364        LogicalTypeKind::Time { precision } => {
365            let value = read_ordered(encoded, 8)? as i64;
366            validate_time(value, *precision)?;
367            Ok((Value::Time(value), 8))
368        }
369        LogicalTypeKind::Timestamp {
370            precision,
371            timestamp_kind,
372        } => {
373            require_len(encoded, 12)?;
374            let seconds = read_ordered(encoded, 8)? as i64;
375            let nanos = u32::from_be_bytes(encoded[8..12].try_into().unwrap());
376            validate_timestamp(
377                *precision,
378                *timestamp_kind,
379                *precision,
380                *timestamp_kind,
381                seconds,
382                nanos,
383            )?;
384            Ok((
385                Value::Timestamp {
386                    precision: *precision,
387                    timestamp_kind: *timestamp_kind,
388                    seconds,
389                    nanos,
390                },
391                12,
392            ))
393        }
394        LogicalTypeKind::String => {
395            let (bytes, consumed) = read_escaped(encoded)?;
396            let value =
397                String::from_utf8(bytes).map_err(|_| TableError::codec("invalid UTF-8 key"))?;
398            Ok((Value::String(value), consumed))
399        }
400        LogicalTypeKind::Binary => {
401            let (bytes, consumed) = read_escaped(encoded)?;
402            Ok((Value::Binary(Bytes::from(bytes)), consumed))
403        }
404        _ => Err(TableError::codec("type cannot be used as a key")),
405    }
406}
407
408fn encode_value(logical_type: &LogicalType, value: &Value, out: &mut Vec<u8>) -> Result<()> {
409    if logical_type.nullable {
410        match value {
411            Value::Null => {
412                out.push(0);
413                return Ok(());
414            }
415            _ => out.push(1),
416        }
417    } else if matches!(value, Value::Null) {
418        return Err(type_mismatch(logical_type, value));
419    }
420    encode_non_null(logical_type, value, out)
421}
422
423fn encode_non_null(logical_type: &LogicalType, value: &Value, out: &mut Vec<u8>) -> Result<()> {
424    match (&logical_type.kind, value) {
425        (LogicalTypeKind::Boolean, Value::Boolean(value)) => out.push(u8::from(*value)),
426        (LogicalTypeKind::Int8, Value::Int8(value)) => ordered_i128(*value as i128, 1, out),
427        (LogicalTypeKind::Int16, Value::Int16(value)) => ordered_i128(*value as i128, 2, out),
428        (LogicalTypeKind::Int32, Value::Int32(value)) => ordered_i128(*value as i128, 4, out),
429        (LogicalTypeKind::Int64, Value::Int64(value)) => ordered_i128(*value as i128, 8, out),
430        (LogicalTypeKind::Float32, Value::Float32(value)) => {
431            out.extend_from_slice(&value.to_le_bytes())
432        }
433        (LogicalTypeKind::Float64, Value::Float64(value)) => {
434            out.extend_from_slice(&value.to_le_bytes())
435        }
436        (
437            LogicalTypeKind::Decimal { precision, scale },
438            Value::Decimal {
439                precision: actual_precision,
440                scale: actual_scale,
441                unscaled,
442            },
443        ) => {
444            validate_decimal(
445                *precision,
446                *scale,
447                *actual_precision,
448                *actual_scale,
449                *unscaled,
450            )?;
451            ordered_i128(*unscaled, decimal_width(*precision), out);
452        }
453        (LogicalTypeKind::Date, Value::Date(value)) => ordered_i128(*value as i128, 4, out),
454        (LogicalTypeKind::Time { precision }, Value::Time(value)) => {
455            validate_time(*value, *precision)?;
456            ordered_i128(*value as i128, 8, out);
457        }
458        (
459            LogicalTypeKind::Timestamp {
460                precision,
461                timestamp_kind,
462            },
463            Value::Timestamp {
464                precision: actual_precision,
465                timestamp_kind: actual_kind,
466                seconds,
467                nanos,
468            },
469        ) => {
470            validate_timestamp(
471                *precision,
472                *timestamp_kind,
473                *actual_precision,
474                *actual_kind,
475                *seconds,
476                *nanos,
477            )?;
478            ordered_i128(*seconds as i128, 8, out);
479            out.extend_from_slice(&nanos.to_be_bytes());
480        }
481        (LogicalTypeKind::String, Value::String(value)) => out.extend_from_slice(value.as_bytes()),
482        (LogicalTypeKind::Binary, Value::Binary(value)) => out.extend_from_slice(value),
483        (LogicalTypeKind::List { element_type }, Value::List(values)) => {
484            append_count(values.len(), out)?;
485            for value in values {
486                append_framed(element_type, value, out)?;
487            }
488        }
489        (
490            LogicalTypeKind::Map {
491                key_type,
492                value_type,
493            },
494            Value::Map(entries),
495        ) => {
496            append_count(entries.len(), out)?;
497            for (key, value) in entries {
498                append_framed(key_type, key, out)?;
499                append_framed(value_type, value, out)?;
500            }
501        }
502        (LogicalTypeKind::Struct { fields }, Value::Struct(values)) => {
503            if fields.len() != values.len() {
504                return Err(TableError::codec(
505                    "struct field count does not match schema",
506                ));
507            }
508            for (field, value) in fields.iter().zip(values) {
509                append_framed(&field.logical_type, value, out)?;
510            }
511        }
512        (LogicalTypeKind::Extension { extension }, Value::Extension { type_id, value })
513            if type_id == &extension.type_id =>
514        {
515            encode_value(&extension.physical_type, value, out)?;
516        }
517        _ => return Err(type_mismatch(logical_type, value)),
518    }
519    Ok(())
520}
521
522fn decode_value(logical_type: &LogicalType, input: Input<'_>) -> Result<Value> {
523    if !logical_type.nullable {
524        return decode_non_null(logical_type, input);
525    }
526    let bytes = input.bytes();
527    let Some((&marker, payload)) = bytes.split_first() else {
528        return Err(TableError::codec("nullable value is missing marker"));
529    };
530    match marker {
531        0 if payload.is_empty() => Ok(Value::Null),
532        0 => Err(TableError::codec("null value has trailing bytes")),
533        1 => decode_non_null(logical_type, input.slice(1..bytes.len())),
534        _ => Err(TableError::codec("invalid nullable marker")),
535    }
536}
537
538fn decode_non_null(logical_type: &LogicalType, input: Input<'_>) -> Result<Value> {
539    let bytes = input.bytes();
540    match &logical_type.kind {
541        LogicalTypeKind::Boolean => Ok(Value::Boolean(read_bool_exact(bytes)?)),
542        LogicalTypeKind::Int8 => Ok(Value::Int8(read_ordered_exact(bytes, 1)? as i8)),
543        LogicalTypeKind::Int16 => Ok(Value::Int16(read_ordered_exact(bytes, 2)? as i16)),
544        LogicalTypeKind::Int32 => Ok(Value::Int32(read_ordered_exact(bytes, 4)? as i32)),
545        LogicalTypeKind::Int64 => Ok(Value::Int64(read_ordered_exact(bytes, 8)? as i64)),
546        LogicalTypeKind::Float32 => Ok(Value::Float32(f32::from_le_bytes(read_exact(bytes)?))),
547        LogicalTypeKind::Float64 => Ok(Value::Float64(f64::from_le_bytes(read_exact(bytes)?))),
548        LogicalTypeKind::Decimal { precision, scale } => {
549            let width = decimal_width(*precision);
550            require_exact_len(bytes, width)?;
551            let unscaled = read_ordered(bytes, width)?;
552            validate_decimal(*precision, *scale, *precision, *scale, unscaled)?;
553            Ok(Value::Decimal {
554                precision: *precision,
555                scale: *scale,
556                unscaled,
557            })
558        }
559        LogicalTypeKind::Date => Ok(Value::Date(read_ordered_exact(bytes, 4)? as i32)),
560        LogicalTypeKind::Time { precision } => {
561            let value = read_ordered_exact(bytes, 8)? as i64;
562            validate_time(value, *precision)?;
563            Ok(Value::Time(value))
564        }
565        LogicalTypeKind::Timestamp {
566            precision,
567            timestamp_kind,
568        } => {
569            require_exact_len(bytes, 12)?;
570            let seconds = read_ordered(bytes, 8)? as i64;
571            let nanos = u32::from_be_bytes(bytes[8..12].try_into().unwrap());
572            validate_timestamp(
573                *precision,
574                *timestamp_kind,
575                *precision,
576                *timestamp_kind,
577                seconds,
578                nanos,
579            )?;
580            Ok(Value::Timestamp {
581                precision: *precision,
582                timestamp_kind: *timestamp_kind,
583                seconds,
584                nanos,
585            })
586        }
587        LogicalTypeKind::String => String::from_utf8(bytes.to_vec())
588            .map(Value::String)
589            .map_err(|_| TableError::codec("invalid UTF-8 value")),
590        LogicalTypeKind::Binary => Ok(Value::Binary(input.into_bytes())),
591        LogicalTypeKind::List { element_type } => {
592            let mut cursor = 0;
593            let count = read_count(bytes, &mut cursor)?;
594            if count > (bytes.len() - cursor) / 4 {
595                return Err(TableError::codec("list count exceeds encoded frames"));
596            }
597            let mut values = Vec::with_capacity(count);
598            for _ in 0..count {
599                values.push(decode_framed(element_type, &input, &mut cursor)?);
600            }
601            ensure_consumed(bytes.len(), cursor)?;
602            Ok(Value::List(values))
603        }
604        LogicalTypeKind::Map {
605            key_type,
606            value_type,
607        } => {
608            let mut cursor = 0;
609            let count = read_count(bytes, &mut cursor)?;
610            if count > (bytes.len() - cursor) / 8 {
611                return Err(TableError::codec("map count exceeds encoded frames"));
612            }
613            let mut entries = Vec::with_capacity(count);
614            for _ in 0..count {
615                let key = decode_framed(key_type, &input, &mut cursor)?;
616                let value = decode_framed(value_type, &input, &mut cursor)?;
617                entries.push((key, value));
618            }
619            ensure_consumed(bytes.len(), cursor)?;
620            Ok(Value::Map(entries))
621        }
622        LogicalTypeKind::Struct { fields } => {
623            let mut cursor = 0;
624            let mut values = Vec::with_capacity(fields.len());
625            for field in fields {
626                values.push(decode_framed(&field.logical_type, &input, &mut cursor)?);
627            }
628            ensure_consumed(bytes.len(), cursor)?;
629            Ok(Value::Struct(values))
630        }
631        LogicalTypeKind::Extension { extension } => Ok(Value::Extension {
632            type_id: extension.type_id.clone(),
633            value: Box::new(decode_value(&extension.physical_type, input)?),
634        }),
635    }
636}
637
638fn value_size(logical_type: &LogicalType, value: &Value) -> Result<usize> {
639    if logical_type.nullable && matches!(value, Value::Null) {
640        return Ok(1);
641    }
642    if !logical_type.nullable && matches!(value, Value::Null) {
643        return Err(type_mismatch(logical_type, value));
644    }
645    let marker = usize::from(logical_type.nullable);
646    let payload = match (&logical_type.kind, value) {
647        (LogicalTypeKind::Boolean, Value::Boolean(_)) | (LogicalTypeKind::Int8, Value::Int8(_)) => {
648            1
649        }
650        (LogicalTypeKind::Int16, Value::Int16(_)) => 2,
651        (LogicalTypeKind::Int32, Value::Int32(_))
652        | (LogicalTypeKind::Float32, Value::Float32(_))
653        | (LogicalTypeKind::Date, Value::Date(_)) => 4,
654        (LogicalTypeKind::Int64, Value::Int64(_))
655        | (LogicalTypeKind::Float64, Value::Float64(_)) => 8,
656        (LogicalTypeKind::Time { precision }, Value::Time(value)) => {
657            validate_time(*value, *precision)?;
658            8
659        }
660        (
661            LogicalTypeKind::Timestamp {
662                precision,
663                timestamp_kind,
664            },
665            Value::Timestamp {
666                precision: actual_precision,
667                timestamp_kind: actual_kind,
668                seconds,
669                nanos,
670            },
671        ) => {
672            validate_timestamp(
673                *precision,
674                *timestamp_kind,
675                *actual_precision,
676                *actual_kind,
677                *seconds,
678                *nanos,
679            )?;
680            12
681        }
682        (
683            LogicalTypeKind::Decimal { precision, scale },
684            Value::Decimal {
685                precision: actual_precision,
686                scale: actual_scale,
687                unscaled,
688            },
689        ) => {
690            validate_decimal(
691                *precision,
692                *scale,
693                *actual_precision,
694                *actual_scale,
695                *unscaled,
696            )?;
697            decimal_width(*precision)
698        }
699        (LogicalTypeKind::String, Value::String(value)) => value.len(),
700        (LogicalTypeKind::Binary, Value::Binary(value)) => value.len(),
701        (LogicalTypeKind::List { element_type }, Value::List(values)) => {
702            ensure_count(values.len())?;
703            values.iter().try_fold(4, |size, value| {
704                checked_size_add(size, framed_size(element_type, value)?)
705            })?
706        }
707        (
708            LogicalTypeKind::Map {
709                key_type,
710                value_type,
711            },
712            Value::Map(entries),
713        ) => {
714            ensure_count(entries.len())?;
715            entries.iter().try_fold(4, |size, (key, value)| {
716                checked_size_add(
717                    size,
718                    checked_size_add(framed_size(key_type, key)?, framed_size(value_type, value)?)?,
719                )
720            })?
721        }
722        (LogicalTypeKind::Struct { fields }, Value::Struct(values))
723            if fields.len() == values.len() =>
724        {
725            fields
726                .iter()
727                .zip(values)
728                .try_fold(0, |size, (field, value)| {
729                    checked_size_add(size, framed_size(&field.logical_type, value)?)
730                })?
731        }
732        (LogicalTypeKind::Extension { extension }, Value::Extension { type_id, value })
733            if type_id == &extension.type_id =>
734        {
735            value_size(&extension.physical_type, value)?
736        }
737        _ => return Err(type_mismatch(logical_type, value)),
738    };
739    checked_size_add(marker, payload)
740}
741
742fn append_framed(logical_type: &LogicalType, value: &Value, out: &mut Vec<u8>) -> Result<()> {
743    let start = out.len();
744    out.extend_from_slice(&[0; 4]);
745    if let Err(error) = encode_value(logical_type, value, out) {
746        out.truncate(start);
747        return Err(error);
748    }
749    let length = match u32::try_from(out.len() - start - 4) {
750        Ok(length) => length,
751        Err(_) => {
752            out.truncate(start);
753            return Err(TableError::codec("nested value exceeds u32 length"));
754        }
755    };
756    out[start..start + 4].copy_from_slice(&length.to_le_bytes());
757    Ok(())
758}
759
760fn decode_framed(
761    logical_type: &LogicalType,
762    input: &Input<'_>,
763    cursor: &mut usize,
764) -> Result<Value> {
765    let bytes = input.bytes();
766    let length = read_u32(bytes, cursor)? as usize;
767    let end = cursor
768        .checked_add(length)
769        .filter(|end| *end <= bytes.len())
770        .ok_or_else(|| TableError::codec("nested value length exceeds container"))?;
771    let value = decode_value(logical_type, input.slice(*cursor..end))?;
772    *cursor = end;
773    Ok(value)
774}
775
776fn append_count(count: usize, out: &mut Vec<u8>) -> Result<()> {
777    out.extend_from_slice(
778        &u32::try_from(count)
779            .map_err(|_| TableError::codec("container has more than u32 values"))?
780            .to_le_bytes(),
781    );
782    Ok(())
783}
784
785fn read_count(bytes: &[u8], cursor: &mut usize) -> Result<usize> {
786    Ok(read_u32(bytes, cursor)? as usize)
787}
788
789fn read_u32(bytes: &[u8], cursor: &mut usize) -> Result<u32> {
790    let end = cursor
791        .checked_add(4)
792        .filter(|end| *end <= bytes.len())
793        .ok_or_else(|| TableError::codec("container is truncated"))?;
794    let value = u32::from_le_bytes(bytes[*cursor..end].try_into().unwrap());
795    *cursor = end;
796    Ok(value)
797}
798
799fn ordered_i128(value: i128, width: usize, out: &mut Vec<u8>) {
800    let start = out.len();
801    out.extend_from_slice(&value.to_be_bytes()[16 - width..]);
802    out[start] ^= 0x80;
803}
804
805fn read_ordered(bytes: &[u8], width: usize) -> Result<i128> {
806    require_len(bytes, width)?;
807    let mut fixed = [0u8; 16];
808    fixed[16 - width..].copy_from_slice(&bytes[..width]);
809    fixed[16 - width] ^= 0x80;
810    if fixed[16 - width] & 0x80 != 0 {
811        fixed[..16 - width].fill(0xff);
812    }
813    Ok(i128::from_be_bytes(fixed))
814}
815
816fn read_ordered_exact(bytes: &[u8], width: usize) -> Result<i128> {
817    require_exact_len(bytes, width)?;
818    read_ordered(bytes, width)
819}
820
821fn append_escaped(bytes: &[u8], out: &mut Vec<u8>) {
822    for byte in bytes {
823        if *byte == 0 {
824            out.extend_from_slice(&[0, 0xff]);
825        } else {
826            out.push(*byte);
827        }
828    }
829    out.extend_from_slice(&[0, 0]);
830}
831
832fn read_escaped(bytes: &[u8]) -> Result<(Vec<u8>, usize)> {
833    let mut out = Vec::new();
834    let mut cursor = 0;
835    while cursor < bytes.len() {
836        if bytes[cursor] != 0 {
837            out.push(bytes[cursor]);
838            cursor += 1;
839            continue;
840        }
841        let next = *bytes
842            .get(cursor + 1)
843            .ok_or_else(|| TableError::codec("unterminated escaped key"))?;
844        match next {
845            0 => return Ok((out, cursor + 2)),
846            0xff => out.push(0),
847            _ => return Err(TableError::codec("invalid key escape")),
848        }
849        cursor += 2;
850    }
851    Err(TableError::codec("unterminated escaped key"))
852}
853
854fn validate_decimal(
855    precision: u8,
856    scale: u8,
857    actual_precision: u8,
858    actual_scale: u8,
859    unscaled: i128,
860) -> Result<()> {
861    if !(1..=38).contains(&precision) || scale > precision {
862        return Err(TableError::codec("invalid decimal value or precision"));
863    }
864    let limit = 10_i128.pow(precision as u32);
865    if precision != actual_precision
866        || scale != actual_scale
867        || unscaled <= -limit
868        || unscaled >= limit
869    {
870        return Err(TableError::codec("invalid decimal value or precision"));
871    }
872    Ok(())
873}
874
875fn decimal_width(precision: u8) -> usize {
876    if precision <= 9 {
877        4
878    } else if precision <= 18 {
879        8
880    } else {
881        16
882    }
883}
884
885fn validate_time(value: i64, precision: u8) -> Result<()> {
886    if precision > 9
887        || !(0..NANOS_PER_DAY).contains(&value)
888        || !(value as u64).is_multiple_of(10_u64.pow((9 - precision) as u32))
889    {
890        return Err(TableError::codec("invalid time value"));
891    }
892    Ok(())
893}
894
895fn validate_timestamp(
896    precision: u8,
897    expected_kind: TimestampKind,
898    actual_precision: u8,
899    actual_kind: TimestampKind,
900    seconds: i64,
901    nanos: u32,
902) -> Result<()> {
903    if precision > 9
904        || precision != actual_precision
905        || expected_kind != actual_kind
906        || nanos >= 1_000_000_000
907        || !nanos.is_multiple_of(10_u32.pow((9 - precision) as u32))
908    {
909        return Err(TableError::codec(format!(
910            "invalid timestamp seconds={seconds} nanos={nanos}"
911        )));
912    }
913    Ok(())
914}
915
916fn read_bool(bytes: &[u8]) -> Result<bool> {
917    match *bytes
918        .first()
919        .ok_or_else(|| TableError::codec("value is truncated"))?
920    {
921        0 => Ok(false),
922        1 => Ok(true),
923        _ => Err(TableError::codec("invalid boolean byte")),
924    }
925}
926fn read_bool_exact(bytes: &[u8]) -> Result<bool> {
927    require_exact_len(bytes, 1)?;
928    read_bool(bytes)
929}
930fn require_len(bytes: &[u8], length: usize) -> Result<()> {
931    if bytes.len() < length {
932        Err(TableError::codec("value is truncated"))
933    } else {
934        Ok(())
935    }
936}
937fn require_exact_len(bytes: &[u8], length: usize) -> Result<()> {
938    if bytes.len() != length {
939        Err(TableError::codec("invalid value length"))
940    } else {
941        Ok(())
942    }
943}
944fn ensure_consumed(length: usize, consumed: usize) -> Result<()> {
945    if length == consumed {
946        Ok(())
947    } else {
948        Err(TableError::codec("trailing bytes"))
949    }
950}
951fn read_exact<const N: usize>(bytes: &[u8]) -> Result<[u8; N]> {
952    require_exact_len(bytes, N)?;
953    Ok(bytes.try_into().unwrap())
954}
955fn type_mismatch(logical_type: &LogicalType, _: &Value) -> TableError {
956    TableError::codec(format!("value does not match {:?}", logical_type.kind))
957}
958
959fn checked_size_add(left: usize, right: usize) -> Result<usize> {
960    left.checked_add(right)
961        .ok_or_else(|| TableError::codec("encoded value exceeds addressable memory"))
962}
963
964fn ensure_count(count: usize) -> Result<()> {
965    u32::try_from(count)
966        .map(|_| ())
967        .map_err(|_| TableError::codec("container has more than u32 values"))
968}
969
970fn framed_size(logical_type: &LogicalType, value: &Value) -> Result<usize> {
971    let payload = value_size(logical_type, value)?;
972    u32::try_from(payload).map_err(|_| TableError::codec("nested value exceeds u32 length"))?;
973    checked_size_add(4, payload)
974}
cobble_table/codec.rs

cobble_table/
codec.rs
