alopex_sql/storage/
codec.rs

1use std::convert::TryFrom;
2
3use crate::planner::ResolvedType;
4
5use super::error::{Result, StorageError};
6use super::value::SqlValue;
7
8const MAX_INLINE_BYTES: usize = 16 * 1024 * 1024; // 16 MiB guard for Text/Blob payloads
9const MAX_VECTOR_LEN: usize = 4 * 1024 * 1024; // 4 million elements (~16 MiB of f32)
10const MAX_NESTED_ELEMENTS: usize = 100_000;
11const MAX_NESTED_DEPTH: usize = 16;
12
13/// RowCodec converts between `Vec<SqlValue>` and a binary TLV format with a null bitmap.
14///
15/// Format:
16/// ```text
17/// [column_count: u16 LE]
18/// [null_bitmap: ceil(count/8) bytes] // bit=1 means NULL
19/// for each non-null column:
20///     [type_tag: u8]
21///     [value_bytes: variable length]
22/// ```
23pub struct RowCodec;
24
25impl RowCodec {
26    /// Encode a row into binary form.
27    pub fn encode(row: &[SqlValue]) -> Vec<u8> {
28        let column_count =
29            u16::try_from(row.len()).expect("row column count exceeds u16::MAX (design limit)");
30        let null_bytes = (column_count as usize).div_ceil(8);
31
32        // Pre-allocate roughly: header + bitmap + average 8 bytes per column.
33        let mut buf = Vec::with_capacity(2 + null_bytes + row.len() * 8);
34        buf.extend_from_slice(&column_count.to_le_bytes());
35
36        let mut null_bitmap = vec![0u8; null_bytes];
37        for (idx, val) in row.iter().enumerate() {
38            if val.is_null() {
39                null_bitmap[idx / 8] |= 1 << (idx % 8);
40            }
41        }
42        buf.extend_from_slice(&null_bitmap);
43
44        for value in row {
45            if value.is_null() {
46                continue;
47            }
48            buf.push(value.type_tag());
49            encode_value(value, &mut buf);
50        }
51
52        buf
53    }
54
55    /// Decode a row from binary form.
56    pub fn decode(bytes: &[u8]) -> Result<Vec<SqlValue>> {
57        let mut cursor = 0;
58        if bytes.len() < 2 {
59            return Err(StorageError::CorruptedData {
60                reason: "missing column count".into(),
61            });
62        }
63
64        let column_count =
65            u16::from_le_bytes(bytes[cursor..cursor + 2].try_into().unwrap()) as usize;
66        cursor += 2;
67
68        let null_bytes = column_count.div_ceil(8);
69        if bytes.len() < cursor + null_bytes {
70            return Err(StorageError::CorruptedData {
71                reason: "missing null bitmap".into(),
72            });
73        }
74        let null_bitmap = &bytes[cursor..cursor + null_bytes];
75        cursor += null_bytes;
76
77        let mut values = Vec::with_capacity(column_count);
78        for idx in 0..column_count {
79            let is_null = (null_bitmap[idx / 8] & (1 << (idx % 8))) != 0;
80            if is_null {
81                values.push(SqlValue::Null);
82                continue;
83            }
84
85            if cursor >= bytes.len() {
86                return Err(StorageError::CorruptedData {
87                    reason: "missing type tag".into(),
88                });
89            }
90            let tag = bytes[cursor];
91            cursor += 1;
92
93            let value = decode_value(tag, bytes, &mut cursor)?;
94            values.push(value);
95        }
96
97        if cursor != bytes.len() {
98            return Err(StorageError::CorruptedData {
99                reason: "trailing bytes after decoding row".into(),
100            });
101        }
102
103        Ok(values)
104    }
105
106    /// Decode with schema validation.
107    pub fn decode_with_schema(bytes: &[u8], schema: &[ResolvedType]) -> Result<Vec<SqlValue>> {
108        let values = Self::decode(bytes)?;
109
110        if values.len() != schema.len() {
111            return Err(StorageError::CorruptedData {
112                reason: format!(
113                    "column count mismatch: encoded={}, expected={}",
114                    values.len(),
115                    schema.len()
116                ),
117            });
118        }
119
120        values
121            .into_iter()
122            .zip(schema.iter())
123            .map(|(value, ty)| ensure_type(value, ty))
124            .collect()
125    }
126}
127
128fn encode_value(value: &SqlValue, buf: &mut Vec<u8>) {
129    match value {
130        SqlValue::Null => {}
131        SqlValue::Integer(v) => buf.extend_from_slice(&v.to_le_bytes()),
132        SqlValue::BigInt(v) => buf.extend_from_slice(&v.to_le_bytes()),
133        SqlValue::Float(v) => buf.extend_from_slice(&v.to_bits().to_le_bytes()),
134        SqlValue::Double(v) => buf.extend_from_slice(&v.to_bits().to_le_bytes()),
135        SqlValue::Text(s) => {
136            let len = u32::try_from(s.len())
137                .expect("text length exceeds u32::MAX (design limit for row encoding)");
138            buf.extend_from_slice(&len.to_le_bytes());
139            buf.extend_from_slice(s.as_bytes());
140        }
141        SqlValue::Blob(bytes) => {
142            let len = u32::try_from(bytes.len())
143                .expect("blob length exceeds u32::MAX (design limit for row encoding)");
144            buf.extend_from_slice(&len.to_le_bytes());
145            buf.extend_from_slice(bytes);
146        }
147        SqlValue::Boolean(b) => buf.push(u8::from(*b)),
148        SqlValue::Timestamp(v) => buf.extend_from_slice(&v.to_le_bytes()),
149        SqlValue::Vector(values) => {
150            let len = u32::try_from(values.len())
151                .expect("vector length exceeds u32::MAX (design limit for row encoding)");
152            buf.extend_from_slice(&len.to_le_bytes());
153            for f in values {
154                buf.extend_from_slice(&f.to_bits().to_le_bytes());
155            }
156        }
157        SqlValue::Date(v) => buf.extend_from_slice(&v.to_le_bytes()),
158        SqlValue::Time(v) => buf.extend_from_slice(&v.to_le_bytes()),
159        SqlValue::Interval {
160            months,
161            days,
162            micros,
163        } => {
164            buf.extend_from_slice(&months.to_le_bytes());
165            buf.extend_from_slice(&days.to_le_bytes());
166            buf.extend_from_slice(&micros.to_le_bytes());
167        }
168        SqlValue::Decimal(value) => {
169            buf.extend_from_slice(&value.coefficient.to_le_bytes());
170            buf.push(value.scale);
171        }
172        SqlValue::Json(value) => {
173            let bytes = value.as_str().as_bytes();
174            let len = u32::try_from(bytes.len())
175                .expect("JSON length exceeds u32::MAX (design limit for row encoding)");
176            buf.extend_from_slice(&len.to_le_bytes());
177            buf.extend_from_slice(bytes);
178        }
179        SqlValue::Array(values) => {
180            buf.extend_from_slice(&(values.len() as u32).to_le_bytes());
181            for value in values {
182                encode_nested(value, buf);
183            }
184        }
185        SqlValue::Map(values) => {
186            buf.extend_from_slice(&(values.len() as u32).to_le_bytes());
187            for (key, value) in values {
188                encode_nested(key, buf);
189                encode_nested(value, buf);
190            }
191        }
192        SqlValue::Struct(values) => {
193            buf.extend_from_slice(&(values.len() as u32).to_le_bytes());
194            for (name, value) in values {
195                buf.extend_from_slice(&(name.len() as u32).to_le_bytes());
196                buf.extend_from_slice(name.as_bytes());
197                encode_nested(value, buf);
198            }
199        }
200    }
201}
202
203fn decode_value(tag: u8, bytes: &[u8], cursor: &mut usize) -> Result<SqlValue> {
204    decode_value_depth(tag, bytes, cursor, 0)
205}
206
207fn encode_nested(value: &SqlValue, buf: &mut Vec<u8>) {
208    let mut encoded = vec![value.type_tag()];
209    encode_value(value, &mut encoded);
210    buf.extend_from_slice(&(encoded.len() as u32).to_le_bytes());
211    buf.extend_from_slice(&encoded);
212}
213
214fn decode_nested(bytes: &[u8], depth: usize) -> Result<SqlValue> {
215    let Some((&tag, payload)) = bytes.split_first() else {
216        return Err(StorageError::CorruptedData {
217            reason: "missing nested type tag".into(),
218        });
219    };
220    let mut cursor = 0;
221    let value = decode_value_depth(tag, payload, &mut cursor, depth)?;
222    if cursor != payload.len() {
223        return Err(StorageError::CorruptedData {
224            reason: "trailing nested value bytes".into(),
225        });
226    }
227    Ok(value)
228}
229
230fn decode_value_depth(tag: u8, bytes: &[u8], cursor: &mut usize, depth: usize) -> Result<SqlValue> {
231    if depth > MAX_NESTED_DEPTH {
232        return Err(StorageError::CorruptedData {
233            reason: "nested value exceeds depth 16".into(),
234        });
235    }
236    let mut take = |len: usize, reason: &'static str| -> Result<&[u8]> {
237        let end = cursor
238            .checked_add(len)
239            .ok_or_else(|| StorageError::CorruptedData {
240                reason: reason.to_string(),
241            })?;
242        if end > bytes.len() {
243            return Err(StorageError::CorruptedData {
244                reason: reason.to_string(),
245            });
246        }
247        let slice = &bytes[*cursor..end];
248        *cursor = end;
249        Ok(slice)
250    };
251
252    match tag {
253        0x00 => Ok(SqlValue::Null),
254        0x01 => {
255            let raw = take(4, "truncated Integer value")?;
256            Ok(SqlValue::Integer(i32::from_le_bytes(
257                raw.try_into().unwrap(),
258            )))
259        }
260        0x02 => {
261            let raw = take(8, "truncated BigInt value")?;
262            Ok(SqlValue::BigInt(i64::from_le_bytes(
263                raw.try_into().unwrap(),
264            )))
265        }
266        0x03 => {
267            let raw = take(4, "truncated Float value")?;
268            Ok(SqlValue::Float(f32::from_bits(u32::from_le_bytes(
269                raw.try_into().unwrap(),
270            ))))
271        }
272        0x04 => {
273            let raw = take(8, "truncated Double value")?;
274            Ok(SqlValue::Double(f64::from_bits(u64::from_le_bytes(
275                raw.try_into().unwrap(),
276            ))))
277        }
278        0x05 => {
279            let len_bytes = take(4, "truncated Text length")?;
280            let len = u32::from_le_bytes(len_bytes.try_into().unwrap()) as usize;
281            if len > MAX_INLINE_BYTES {
282                return Err(StorageError::CorruptedData {
283                    reason: format!("text length exceeds limit: {len}"),
284                });
285            }
286            let raw = take(len, "truncated Text payload")?;
287            let s = String::from_utf8(raw.to_vec()).map_err(|_| StorageError::CorruptedData {
288                reason: "invalid UTF-8 in Text".into(),
289            })?;
290            Ok(SqlValue::Text(s))
291        }
292        0x06 => {
293            let len_bytes = take(4, "truncated Blob length")?;
294            let len = u32::from_le_bytes(len_bytes.try_into().unwrap()) as usize;
295            if len > MAX_INLINE_BYTES {
296                return Err(StorageError::CorruptedData {
297                    reason: format!("blob length exceeds limit: {len}"),
298                });
299            }
300            let raw = take(len, "truncated Blob payload")?;
301            Ok(SqlValue::Blob(raw.to_vec()))
302        }
303        0x07 => {
304            let raw = take(1, "truncated Boolean")?[0];
305            match raw {
306                0 => Ok(SqlValue::Boolean(false)),
307                1 => Ok(SqlValue::Boolean(true)),
308                other => Err(StorageError::CorruptedData {
309                    reason: format!("invalid boolean value: {}", other),
310                }),
311            }
312        }
313        0x08 => {
314            let raw = take(8, "truncated Timestamp value")?;
315            Ok(SqlValue::Timestamp(i64::from_le_bytes(
316                raw.try_into().unwrap(),
317            )))
318        }
319        0x09 => {
320            let len_bytes = take(4, "truncated Vector length")?;
321            let len = u32::from_le_bytes(len_bytes.try_into().unwrap()) as usize;
322            if len > MAX_VECTOR_LEN {
323                return Err(StorageError::CorruptedData {
324                    reason: format!("vector length exceeds limit: {len}"),
325                });
326            }
327            let total = len
328                .checked_mul(4)
329                .ok_or_else(|| StorageError::CorruptedData {
330                    reason: "vector length overflow".into(),
331                })?;
332            let raw = take(total, "truncated Vector payload")?;
333
334            let mut values = Vec::with_capacity(len);
335            for chunk in raw.as_chunks::<4>().0 {
336                values.push(f32::from_bits(u32::from_le_bytes(*chunk)));
337            }
338            Ok(SqlValue::Vector(values))
339        }
340        0x0a => Ok(SqlValue::Date(i32::from_le_bytes(
341            take(4, "truncated Date value")?.try_into().unwrap(),
342        ))),
343        0x0b => Ok(SqlValue::Time(i64::from_le_bytes(
344            take(8, "truncated Time value")?.try_into().unwrap(),
345        ))),
346        0x0c => Ok(SqlValue::Interval {
347            months: i32::from_le_bytes(take(4, "truncated Interval months")?.try_into().unwrap()),
348            days: i32::from_le_bytes(take(4, "truncated Interval days")?.try_into().unwrap()),
349            micros: i64::from_le_bytes(take(8, "truncated Interval micros")?.try_into().unwrap()),
350        }),
351        0x0d => Ok(SqlValue::Decimal(super::DecimalValue::new(
352            i128::from_le_bytes(
353                take(16, "truncated Decimal coefficient")?
354                    .try_into()
355                    .unwrap(),
356            ),
357            take(1, "truncated Decimal scale")?[0],
358        ))),
359        0x0e => {
360            let len =
361                u32::from_le_bytes(take(4, "truncated JSON length")?.try_into().unwrap()) as usize;
362            if len > MAX_INLINE_BYTES {
363                return Err(StorageError::CorruptedData {
364                    reason: format!("JSON length exceeds limit: {len}"),
365                });
366            }
367            let text = std::str::from_utf8(take(len, "truncated JSON payload")?).map_err(|_| {
368                StorageError::CorruptedData {
369                    reason: "invalid UTF-8 in JSON".into(),
370                }
371            })?;
372            Ok(SqlValue::Json(super::JsonValue::parse(text).map_err(
373                |_| StorageError::CorruptedData {
374                    reason: "invalid JSON payload".into(),
375                },
376            )?))
377        }
378        0x0f => {
379            let len =
380                u32::from_le_bytes(take(4, "truncated Array length")?.try_into().unwrap()) as usize;
381            if len > MAX_NESTED_ELEMENTS {
382                return Err(StorageError::CorruptedData {
383                    reason: format!("array length exceeds limit: {len}"),
384                });
385            }
386            let mut values = Vec::with_capacity(len);
387            for _ in 0..len {
388                let size = u32::from_le_bytes(
389                    take(4, "truncated nested value length")?
390                        .try_into()
391                        .unwrap(),
392                ) as usize;
393                values.push(decode_nested(
394                    take(size, "truncated nested value")?,
395                    depth + 1,
396                )?);
397            }
398            Ok(SqlValue::Array(values))
399        }
400        0x10 => {
401            let len =
402                u32::from_le_bytes(take(4, "truncated Map length")?.try_into().unwrap()) as usize;
403            if len > MAX_NESTED_ELEMENTS {
404                return Err(StorageError::CorruptedData {
405                    reason: format!("map length exceeds limit: {len}"),
406                });
407            }
408            let mut values = Vec::with_capacity(len);
409            for _ in 0..len {
410                let key_size =
411                    u32::from_le_bytes(take(4, "truncated map key length")?.try_into().unwrap())
412                        as usize;
413                let key = decode_nested(take(key_size, "truncated map key")?, depth + 1)?;
414                let value_size =
415                    u32::from_le_bytes(take(4, "truncated map value length")?.try_into().unwrap())
416                        as usize;
417                let value = decode_nested(take(value_size, "truncated map value")?, depth + 1)?;
418                values.push((key, value));
419            }
420            Ok(SqlValue::Map(values))
421        }
422        0x11 => {
423            let len = u32::from_le_bytes(take(4, "truncated Struct length")?.try_into().unwrap())
424                as usize;
425            if len > MAX_NESTED_ELEMENTS {
426                return Err(StorageError::CorruptedData {
427                    reason: format!("struct length exceeds limit: {len}"),
428                });
429            }
430            let mut values = Vec::with_capacity(len);
431            for _ in 0..len {
432                let name_len = u32::from_le_bytes(
433                    take(4, "truncated struct field name length")?
434                        .try_into()
435                        .unwrap(),
436                ) as usize;
437                let name = std::str::from_utf8(take(name_len, "truncated struct field name")?)
438                    .map_err(|_| StorageError::CorruptedData {
439                        reason: "invalid UTF-8 in struct field name".into(),
440                    })?
441                    .to_string();
442                let value_size = u32::from_le_bytes(
443                    take(4, "truncated struct field length")?
444                        .try_into()
445                        .unwrap(),
446                ) as usize;
447                let value = decode_nested(take(value_size, "truncated struct field")?, depth + 1)?;
448                values.push((name, value));
449            }
450            Ok(SqlValue::Struct(values))
451        }
452        other => Err(StorageError::CorruptedData {
453            reason: format!("unknown type tag: 0x{other:02x}"),
454        }),
455    }
456}
457
458fn ensure_type(value: SqlValue, expected: &ResolvedType) -> Result<SqlValue> {
459    use ResolvedType::*;
460    match (expected, value) {
461        (_, SqlValue::Null) => Ok(SqlValue::Null),
462        (Integer, SqlValue::Integer(v)) => Ok(SqlValue::Integer(v)),
463        (BigInt, SqlValue::BigInt(v)) => Ok(SqlValue::BigInt(v)),
464        (Float, SqlValue::Float(v)) => Ok(SqlValue::Float(v)),
465        (Double, SqlValue::Double(v)) => Ok(SqlValue::Double(v)),
466        (Text, SqlValue::Text(s)) => Ok(SqlValue::Text(s)),
467        (Blob, SqlValue::Blob(b)) => Ok(SqlValue::Blob(b)),
468        (Boolean, SqlValue::Boolean(v)) => Ok(SqlValue::Boolean(v)),
469        (Timestamp, SqlValue::Timestamp(v)) => Ok(SqlValue::Timestamp(v)),
470        (Date, SqlValue::Date(v)) => Ok(SqlValue::Date(v)),
471        (Time, SqlValue::Time(v)) => Ok(SqlValue::Time(v)),
472        (Interval, value @ SqlValue::Interval { .. }) => Ok(value),
473        (Decimal { precision, scale }, SqlValue::Decimal(value)) => {
474            let value = value
475                .rescale(*scale)
476                .ok_or_else(|| StorageError::TypeMismatch {
477                    expected: format!("Decimal({precision},{scale})"),
478                    actual: "Decimal overflow".into(),
479                })?;
480            if value.fits_precision(*precision) {
481                Ok(SqlValue::Decimal(value))
482            } else {
483                Err(StorageError::TypeMismatch {
484                    expected: format!("Decimal({precision},{scale})"),
485                    actual: value.to_string(),
486                })
487            }
488        }
489        (Json, SqlValue::Json(value)) => Ok(SqlValue::Json(value)),
490        (Array(element), SqlValue::Array(values)) => values
491            .into_iter()
492            .map(|value| ensure_type(value, element))
493            .collect::<Result<Vec<_>>>()
494            .map(SqlValue::Array),
495        (
496            Map {
497                key: key_type,
498                value: value_type,
499            },
500            SqlValue::Map(values),
501        ) => values
502            .into_iter()
503            .map(|(key, value)| {
504                if key.is_null() {
505                    return Err(StorageError::TypeMismatch {
506                        expected: "non-NULL map key".into(),
507                        actual: "Null".into(),
508                    });
509                }
510                Ok((ensure_type(key, key_type)?, ensure_type(value, value_type)?))
511            })
512            .collect::<Result<Vec<_>>>()
513            .map(SqlValue::Map),
514        (Struct(fields), SqlValue::Struct(values)) if fields.len() == values.len() => values
515            .into_iter()
516            .zip(fields)
517            .map(|((name, value), (expected_name, expected_type))| {
518                if name != *expected_name {
519                    return Err(StorageError::TypeMismatch {
520                        expected: expected_name.clone(),
521                        actual: name,
522                    });
523                }
524                Ok((name, ensure_type(value, expected_type)?))
525            })
526            .collect::<Result<Vec<_>>>()
527            .map(SqlValue::Struct),
528        (Vector { dimension, .. }, SqlValue::Vector(values)) => {
529            if values.len() as u32 == *dimension {
530                Ok(SqlValue::Vector(values))
531            } else {
532                Err(StorageError::TypeMismatch {
533                    expected: format!("Vector(dim={})", dimension),
534                    actual: format!("Vector(dim={})", values.len()),
535                })
536            }
537        }
538        (expected_ty, actual) => Err(StorageError::TypeMismatch {
539            expected: expected_ty.type_name().to_string(),
540            actual: actual.type_name().to_string(),
541        }),
542    }
543}
544
545#[cfg(test)]
546mod tests {
547    use super::*;
548    use proptest::prelude::*;
549
550    fn values_equal(a: &SqlValue, b: &SqlValue) -> bool {
551        match (a, b) {
552            (SqlValue::Float(x), SqlValue::Float(y)) => x.to_bits() == y.to_bits(),
553            (SqlValue::Double(x), SqlValue::Double(y)) => x.to_bits() == y.to_bits(),
554            (SqlValue::Vector(xs), SqlValue::Vector(ys)) => {
555                xs.len() == ys.len()
556                    && xs
557                        .iter()
558                        .zip(ys.iter())
559                        .all(|(x, y)| x.to_bits() == y.to_bits())
560            }
561            _ => a == b,
562        }
563    }
564
565    fn row_equal(a: &[SqlValue], b: &[SqlValue]) -> bool {
566        a.len() == b.len()
567            && a.iter()
568                .zip(b.iter())
569                .all(|(lhs, rhs)| values_equal(lhs, rhs))
570    }
571
572    fn sql_value_strategy() -> impl Strategy<Value = SqlValue> {
573        let finite_f32 = any::<f32>();
574        let finite_f64 = any::<f64>();
575        let decimal_max = crate::storage::value::decimal_power(38).unwrap() - 1;
576        prop_oneof![
577            Just(SqlValue::Null),
578            any::<i32>().prop_map(SqlValue::Integer),
579            any::<i64>().prop_map(SqlValue::BigInt),
580            finite_f32.prop_map(SqlValue::Float),
581            finite_f64.prop_map(SqlValue::Double),
582            ".*".prop_map(SqlValue::Text),
583            proptest::collection::vec(any::<u8>(), 0..32).prop_map(SqlValue::Blob),
584            any::<bool>().prop_map(SqlValue::Boolean),
585            any::<i64>().prop_map(SqlValue::Timestamp),
586            proptest::collection::vec(any::<f32>(), 0..8).prop_map(SqlValue::Vector),
587            any::<i32>().prop_map(SqlValue::Date),
588            any::<i64>().prop_map(SqlValue::Time),
589            (any::<i32>(), any::<i32>(), any::<i64>()).prop_map(|(months, days, micros)| {
590                SqlValue::Interval {
591                    months,
592                    days,
593                    micros,
594                }
595            }),
596            (-decimal_max..=decimal_max, 0_u8..=38).prop_map(|(coefficient, scale)| {
597                SqlValue::Decimal(crate::storage::DecimalValue::new(coefficient, scale))
598            }),
599        ]
600    }
601
602    #[test]
603    fn roundtrip_preserves_all_types() {
604        let row = vec![
605            SqlValue::Null,
606            SqlValue::Integer(42),
607            SqlValue::BigInt(-42),
608            SqlValue::Float(1.5),
609            SqlValue::Double(-2.5),
610            SqlValue::Text("hello".into()),
611            SqlValue::Blob(vec![0x01, 0x02]),
612            SqlValue::Boolean(true),
613            SqlValue::Timestamp(1_700_000_000),
614            SqlValue::Vector(vec![0.1, 0.2, 0.3]),
615            SqlValue::Date(19_782),
616            SqlValue::Time(86_399_123_456),
617            SqlValue::Interval {
618                months: -1,
619                days: 2,
620                micros: 3,
621            },
622            SqlValue::Decimal(crate::storage::DecimalValue::new(-12345, 2)),
623            SqlValue::Array(vec![SqlValue::Integer(1), SqlValue::Null]),
624            SqlValue::Map(vec![(SqlValue::Text("a".into()), SqlValue::Integer(1))]),
625            SqlValue::Struct(vec![(
626                "items".into(),
627                SqlValue::Array(vec![SqlValue::Text("x".into())]),
628            )]),
629        ];
630
631        let encoded = RowCodec::encode(&row);
632        let decoded = RowCodec::decode(&encoded).unwrap();
633
634        assert!(row_equal(&row, &decoded));
635    }
636
637    #[test]
638    fn pre_temporal_row_bytes_remain_readable() {
639        let bytes = [
640            2, 0, // column count
641            0, // null bitmap
642            0x01, 42, 0, 0, 0, // Integer(42)
643            0x08, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, // Timestamp(-1)
644        ];
645
646        assert_eq!(
647            RowCodec::decode(&bytes).unwrap(),
648            vec![SqlValue::Integer(42), SqlValue::Timestamp(-1)]
649        );
650    }
651
652    #[test]
653    fn null_bitmap_is_respected() {
654        let row = vec![SqlValue::Integer(1), SqlValue::Null, SqlValue::Integer(2)];
655        let encoded = RowCodec::encode(&row);
656        let decoded = RowCodec::decode(&encoded).unwrap();
657        assert!(matches!(decoded[1], SqlValue::Null));
658    }
659
660    #[test]
661    fn corruption_is_detected_for_truncated_payload() {
662        let row = vec![SqlValue::Text("abc".into())];
663        let mut encoded = RowCodec::encode(&row);
664        encoded.pop(); // truncate
665        let err = RowCodec::decode(&encoded).unwrap_err();
666        assert!(matches!(err, StorageError::CorruptedData { .. }));
667    }
668
669    #[test]
670    fn corruption_is_detected_for_unknown_tag() {
671        // column_count=1, null_bitmap=0, tag=0xFF (invalid)
672        let bytes = vec![1, 0, 0, 0xFF];
673        let err = RowCodec::decode(&bytes).unwrap_err();
674        assert!(matches!(err, StorageError::CorruptedData { .. }));
675    }
676
677    #[test]
678    fn oversized_lengths_are_rejected() {
679        // Text length = MAX_INLINE_BYTES + 1 with no payload.
680        let mut bytes = Vec::new();
681        bytes.extend_from_slice(&(1u16).to_le_bytes()); // column count
682        bytes.push(0); // null bitmap
683        bytes.push(0x05); // text tag
684        let too_large = (super::MAX_INLINE_BYTES as u32) + 1;
685        bytes.extend_from_slice(&too_large.to_le_bytes());
686        let err = RowCodec::decode(&bytes).unwrap_err();
687        assert!(matches!(err, StorageError::CorruptedData { .. }));
688    }
689
690    #[test]
691    fn oversized_vector_is_rejected() {
692        let mut bytes = Vec::new();
693        bytes.extend_from_slice(&(1u16).to_le_bytes()); // column count
694        bytes.push(0); // null bitmap
695        bytes.push(0x09); // vector tag
696        let too_large = (super::MAX_VECTOR_LEN as u32) + 1;
697        bytes.extend_from_slice(&too_large.to_le_bytes());
698        let err = RowCodec::decode(&bytes).unwrap_err();
699        assert!(matches!(err, StorageError::CorruptedData { .. }));
700    }
701
702    #[test]
703    fn decode_with_schema_validates_types() {
704        let row = vec![SqlValue::Vector(vec![1.0, 2.0])];
705        let encoded = RowCodec::encode(&row);
706        let schema = vec![ResolvedType::Vector {
707            dimension: 3,
708            metric: crate::ast::ddl::VectorMetric::Cosine,
709        }];
710        let err = RowCodec::decode_with_schema(&encoded, &schema).unwrap_err();
711        assert!(matches!(err, StorageError::TypeMismatch { .. }));
712    }
713
714    proptest! {
715        #[test]
716        fn proptest_roundtrip(row in proptest::collection::vec(sql_value_strategy(), 0..16)) {
717            let encoded = RowCodec::encode(&row);
718            let decoded = RowCodec::decode(&encoded).unwrap();
719            prop_assert!(row_equal(&row, &decoded));
720        }
721
722        #[test]
723        fn decode_with_schema_matches_lengths(row in proptest::collection::vec(sql_value_strategy(), 1..5)) {
724            let schema: Vec<ResolvedType> = row.iter().map(|v| v.resolved_type()).collect();
725            let encoded = RowCodec::encode(&row);
726            let decoded = RowCodec::decode_with_schema(&encoded, &schema).unwrap();
727            prop_assert!(row_equal(&row, &decoded));
728        }
729    }
730}
alopex_sql/storage/codec.rs

alopex_sql/storage/
codec.rs
