use std::convert::TryFrom;
use crate::planner::ResolvedType;
use super::error::{Result, StorageError};
use super::value::SqlValue;
const MAX_INLINE_BYTES: usize = 16 * 1024 * 1024; const MAX_VECTOR_LEN: usize = 4 * 1024 * 1024; const MAX_NESTED_ELEMENTS: usize = 100_000;
const MAX_NESTED_DEPTH: usize = 16;
pub struct RowCodec;
impl RowCodec {
pub fn encode(row: &[SqlValue]) -> Vec<u8> {
let column_count =
u16::try_from(row.len()).expect("row column count exceeds u16::MAX (design limit)");
let null_bytes = (column_count as usize).div_ceil(8);
let mut buf = Vec::with_capacity(2 + null_bytes + row.len() * 8);
buf.extend_from_slice(&column_count.to_le_bytes());
let mut null_bitmap = vec![0u8; null_bytes];
for (idx, val) in row.iter().enumerate() {
if val.is_null() {
null_bitmap[idx / 8] |= 1 << (idx % 8);
}
}
buf.extend_from_slice(&null_bitmap);
for value in row {
if value.is_null() {
continue;
}
buf.push(value.type_tag());
encode_value(value, &mut buf);
}
buf
}
pub fn decode(bytes: &[u8]) -> Result<Vec<SqlValue>> {
let mut cursor = 0;
if bytes.len() < 2 {
return Err(StorageError::CorruptedData {
reason: "missing column count".into(),
});
}
let column_count =
u16::from_le_bytes(bytes[cursor..cursor + 2].try_into().unwrap()) as usize;
cursor += 2;
let null_bytes = column_count.div_ceil(8);
if bytes.len() < cursor + null_bytes {
return Err(StorageError::CorruptedData {
reason: "missing null bitmap".into(),
});
}
let null_bitmap = &bytes[cursor..cursor + null_bytes];
cursor += null_bytes;
let mut values = Vec::with_capacity(column_count);
for idx in 0..column_count {
let is_null = (null_bitmap[idx / 8] & (1 << (idx % 8))) != 0;
if is_null {
values.push(SqlValue::Null);
continue;
}
if cursor >= bytes.len() {
return Err(StorageError::CorruptedData {
reason: "missing type tag".into(),
});
}
let tag = bytes[cursor];
cursor += 1;
let value = decode_value(tag, bytes, &mut cursor)?;
values.push(value);
}
if cursor != bytes.len() {
return Err(StorageError::CorruptedData {
reason: "trailing bytes after decoding row".into(),
});
}
Ok(values)
}
pub fn decode_with_schema(bytes: &[u8], schema: &[ResolvedType]) -> Result<Vec<SqlValue>> {
let values = Self::decode(bytes)?;
if values.len() != schema.len() {
return Err(StorageError::CorruptedData {
reason: format!(
"column count mismatch: encoded={}, expected={}",
values.len(),
schema.len()
),
});
}
values
.into_iter()
.zip(schema.iter())
.map(|(value, ty)| ensure_type(value, ty))
.collect()
}
}
fn encode_value(value: &SqlValue, buf: &mut Vec<u8>) {
match value {
SqlValue::Null => {}
SqlValue::Integer(v) => buf.extend_from_slice(&v.to_le_bytes()),
SqlValue::BigInt(v) => buf.extend_from_slice(&v.to_le_bytes()),
SqlValue::Float(v) => buf.extend_from_slice(&v.to_bits().to_le_bytes()),
SqlValue::Double(v) => buf.extend_from_slice(&v.to_bits().to_le_bytes()),
SqlValue::Text(s) => {
let len = u32::try_from(s.len())
.expect("text length exceeds u32::MAX (design limit for row encoding)");
buf.extend_from_slice(&len.to_le_bytes());
buf.extend_from_slice(s.as_bytes());
}
SqlValue::Blob(bytes) => {
let len = u32::try_from(bytes.len())
.expect("blob length exceeds u32::MAX (design limit for row encoding)");
buf.extend_from_slice(&len.to_le_bytes());
buf.extend_from_slice(bytes);
}
SqlValue::Boolean(b) => buf.push(u8::from(*b)),
SqlValue::Timestamp(v) => buf.extend_from_slice(&v.to_le_bytes()),
SqlValue::Vector(values) => {
let len = u32::try_from(values.len())
.expect("vector length exceeds u32::MAX (design limit for row encoding)");
buf.extend_from_slice(&len.to_le_bytes());
for f in values {
buf.extend_from_slice(&f.to_bits().to_le_bytes());
}
}
SqlValue::Date(v) => buf.extend_from_slice(&v.to_le_bytes()),
SqlValue::Time(v) => buf.extend_from_slice(&v.to_le_bytes()),
SqlValue::Interval {
months,
days,
micros,
} => {
buf.extend_from_slice(&months.to_le_bytes());
buf.extend_from_slice(&days.to_le_bytes());
buf.extend_from_slice(µs.to_le_bytes());
}
SqlValue::Decimal(value) => {
buf.extend_from_slice(&value.coefficient.to_le_bytes());
buf.push(value.scale);
}
SqlValue::Json(value) => {
let bytes = value.as_str().as_bytes();
let len = u32::try_from(bytes.len())
.expect("JSON length exceeds u32::MAX (design limit for row encoding)");
buf.extend_from_slice(&len.to_le_bytes());
buf.extend_from_slice(bytes);
}
SqlValue::Array(values) => {
buf.extend_from_slice(&(values.len() as u32).to_le_bytes());
for value in values {
encode_nested(value, buf);
}
}
SqlValue::Map(values) => {
buf.extend_from_slice(&(values.len() as u32).to_le_bytes());
for (key, value) in values {
encode_nested(key, buf);
encode_nested(value, buf);
}
}
SqlValue::Struct(values) => {
buf.extend_from_slice(&(values.len() as u32).to_le_bytes());
for (name, value) in values {
buf.extend_from_slice(&(name.len() as u32).to_le_bytes());
buf.extend_from_slice(name.as_bytes());
encode_nested(value, buf);
}
}
}
}
fn decode_value(tag: u8, bytes: &[u8], cursor: &mut usize) -> Result<SqlValue> {
decode_value_depth(tag, bytes, cursor, 0)
}
fn encode_nested(value: &SqlValue, buf: &mut Vec<u8>) {
let mut encoded = vec![value.type_tag()];
encode_value(value, &mut encoded);
buf.extend_from_slice(&(encoded.len() as u32).to_le_bytes());
buf.extend_from_slice(&encoded);
}
fn decode_nested(bytes: &[u8], depth: usize) -> Result<SqlValue> {
let Some((&tag, payload)) = bytes.split_first() else {
return Err(StorageError::CorruptedData {
reason: "missing nested type tag".into(),
});
};
let mut cursor = 0;
let value = decode_value_depth(tag, payload, &mut cursor, depth)?;
if cursor != payload.len() {
return Err(StorageError::CorruptedData {
reason: "trailing nested value bytes".into(),
});
}
Ok(value)
}
fn decode_value_depth(tag: u8, bytes: &[u8], cursor: &mut usize, depth: usize) -> Result<SqlValue> {
if depth > MAX_NESTED_DEPTH {
return Err(StorageError::CorruptedData {
reason: "nested value exceeds depth 16".into(),
});
}
let mut take = |len: usize, reason: &'static str| -> Result<&[u8]> {
let end = cursor
.checked_add(len)
.ok_or_else(|| StorageError::CorruptedData {
reason: reason.to_string(),
})?;
if end > bytes.len() {
return Err(StorageError::CorruptedData {
reason: reason.to_string(),
});
}
let slice = &bytes[*cursor..end];
*cursor = end;
Ok(slice)
};
match tag {
0x00 => Ok(SqlValue::Null),
0x01 => {
let raw = take(4, "truncated Integer value")?;
Ok(SqlValue::Integer(i32::from_le_bytes(
raw.try_into().unwrap(),
)))
}
0x02 => {
let raw = take(8, "truncated BigInt value")?;
Ok(SqlValue::BigInt(i64::from_le_bytes(
raw.try_into().unwrap(),
)))
}
0x03 => {
let raw = take(4, "truncated Float value")?;
Ok(SqlValue::Float(f32::from_bits(u32::from_le_bytes(
raw.try_into().unwrap(),
))))
}
0x04 => {
let raw = take(8, "truncated Double value")?;
Ok(SqlValue::Double(f64::from_bits(u64::from_le_bytes(
raw.try_into().unwrap(),
))))
}
0x05 => {
let len_bytes = take(4, "truncated Text length")?;
let len = u32::from_le_bytes(len_bytes.try_into().unwrap()) as usize;
if len > MAX_INLINE_BYTES {
return Err(StorageError::CorruptedData {
reason: format!("text length exceeds limit: {len}"),
});
}
let raw = take(len, "truncated Text payload")?;
let s = String::from_utf8(raw.to_vec()).map_err(|_| StorageError::CorruptedData {
reason: "invalid UTF-8 in Text".into(),
})?;
Ok(SqlValue::Text(s))
}
0x06 => {
let len_bytes = take(4, "truncated Blob length")?;
let len = u32::from_le_bytes(len_bytes.try_into().unwrap()) as usize;
if len > MAX_INLINE_BYTES {
return Err(StorageError::CorruptedData {
reason: format!("blob length exceeds limit: {len}"),
});
}
let raw = take(len, "truncated Blob payload")?;
Ok(SqlValue::Blob(raw.to_vec()))
}
0x07 => {
let raw = take(1, "truncated Boolean")?[0];
match raw {
0 => Ok(SqlValue::Boolean(false)),
1 => Ok(SqlValue::Boolean(true)),
other => Err(StorageError::CorruptedData {
reason: format!("invalid boolean value: {}", other),
}),
}
}
0x08 => {
let raw = take(8, "truncated Timestamp value")?;
Ok(SqlValue::Timestamp(i64::from_le_bytes(
raw.try_into().unwrap(),
)))
}
0x09 => {
let len_bytes = take(4, "truncated Vector length")?;
let len = u32::from_le_bytes(len_bytes.try_into().unwrap()) as usize;
if len > MAX_VECTOR_LEN {
return Err(StorageError::CorruptedData {
reason: format!("vector length exceeds limit: {len}"),
});
}
let total = len
.checked_mul(4)
.ok_or_else(|| StorageError::CorruptedData {
reason: "vector length overflow".into(),
})?;
let raw = take(total, "truncated Vector payload")?;
let mut values = Vec::with_capacity(len);
for chunk in raw.as_chunks::<4>().0 {
values.push(f32::from_bits(u32::from_le_bytes(*chunk)));
}
Ok(SqlValue::Vector(values))
}
0x0a => Ok(SqlValue::Date(i32::from_le_bytes(
take(4, "truncated Date value")?.try_into().unwrap(),
))),
0x0b => Ok(SqlValue::Time(i64::from_le_bytes(
take(8, "truncated Time value")?.try_into().unwrap(),
))),
0x0c => Ok(SqlValue::Interval {
months: i32::from_le_bytes(take(4, "truncated Interval months")?.try_into().unwrap()),
days: i32::from_le_bytes(take(4, "truncated Interval days")?.try_into().unwrap()),
micros: i64::from_le_bytes(take(8, "truncated Interval micros")?.try_into().unwrap()),
}),
0x0d => Ok(SqlValue::Decimal(super::DecimalValue::new(
i128::from_le_bytes(
take(16, "truncated Decimal coefficient")?
.try_into()
.unwrap(),
),
take(1, "truncated Decimal scale")?[0],
))),
0x0e => {
let len =
u32::from_le_bytes(take(4, "truncated JSON length")?.try_into().unwrap()) as usize;
if len > MAX_INLINE_BYTES {
return Err(StorageError::CorruptedData {
reason: format!("JSON length exceeds limit: {len}"),
});
}
let text = std::str::from_utf8(take(len, "truncated JSON payload")?).map_err(|_| {
StorageError::CorruptedData {
reason: "invalid UTF-8 in JSON".into(),
}
})?;
Ok(SqlValue::Json(super::JsonValue::parse(text).map_err(
|_| StorageError::CorruptedData {
reason: "invalid JSON payload".into(),
},
)?))
}
0x0f => {
let len =
u32::from_le_bytes(take(4, "truncated Array length")?.try_into().unwrap()) as usize;
if len > MAX_NESTED_ELEMENTS {
return Err(StorageError::CorruptedData {
reason: format!("array length exceeds limit: {len}"),
});
}
let mut values = Vec::with_capacity(len);
for _ in 0..len {
let size = u32::from_le_bytes(
take(4, "truncated nested value length")?
.try_into()
.unwrap(),
) as usize;
values.push(decode_nested(
take(size, "truncated nested value")?,
depth + 1,
)?);
}
Ok(SqlValue::Array(values))
}
0x10 => {
let len =
u32::from_le_bytes(take(4, "truncated Map length")?.try_into().unwrap()) as usize;
if len > MAX_NESTED_ELEMENTS {
return Err(StorageError::CorruptedData {
reason: format!("map length exceeds limit: {len}"),
});
}
let mut values = Vec::with_capacity(len);
for _ in 0..len {
let key_size =
u32::from_le_bytes(take(4, "truncated map key length")?.try_into().unwrap())
as usize;
let key = decode_nested(take(key_size, "truncated map key")?, depth + 1)?;
let value_size =
u32::from_le_bytes(take(4, "truncated map value length")?.try_into().unwrap())
as usize;
let value = decode_nested(take(value_size, "truncated map value")?, depth + 1)?;
values.push((key, value));
}
Ok(SqlValue::Map(values))
}
0x11 => {
let len = u32::from_le_bytes(take(4, "truncated Struct length")?.try_into().unwrap())
as usize;
if len > MAX_NESTED_ELEMENTS {
return Err(StorageError::CorruptedData {
reason: format!("struct length exceeds limit: {len}"),
});
}
let mut values = Vec::with_capacity(len);
for _ in 0..len {
let name_len = u32::from_le_bytes(
take(4, "truncated struct field name length")?
.try_into()
.unwrap(),
) as usize;
let name = std::str::from_utf8(take(name_len, "truncated struct field name")?)
.map_err(|_| StorageError::CorruptedData {
reason: "invalid UTF-8 in struct field name".into(),
})?
.to_string();
let value_size = u32::from_le_bytes(
take(4, "truncated struct field length")?
.try_into()
.unwrap(),
) as usize;
let value = decode_nested(take(value_size, "truncated struct field")?, depth + 1)?;
values.push((name, value));
}
Ok(SqlValue::Struct(values))
}
other => Err(StorageError::CorruptedData {
reason: format!("unknown type tag: 0x{other:02x}"),
}),
}
}
fn ensure_type(value: SqlValue, expected: &ResolvedType) -> Result<SqlValue> {
use ResolvedType::*;
match (expected, value) {
(_, SqlValue::Null) => Ok(SqlValue::Null),
(Integer, SqlValue::Integer(v)) => Ok(SqlValue::Integer(v)),
(BigInt, SqlValue::BigInt(v)) => Ok(SqlValue::BigInt(v)),
(Float, SqlValue::Float(v)) => Ok(SqlValue::Float(v)),
(Double, SqlValue::Double(v)) => Ok(SqlValue::Double(v)),
(Text, SqlValue::Text(s)) => Ok(SqlValue::Text(s)),
(Blob, SqlValue::Blob(b)) => Ok(SqlValue::Blob(b)),
(Boolean, SqlValue::Boolean(v)) => Ok(SqlValue::Boolean(v)),
(Timestamp, SqlValue::Timestamp(v)) => Ok(SqlValue::Timestamp(v)),
(Date, SqlValue::Date(v)) => Ok(SqlValue::Date(v)),
(Time, SqlValue::Time(v)) => Ok(SqlValue::Time(v)),
(Interval, value @ SqlValue::Interval { .. }) => Ok(value),
(Decimal { precision, scale }, SqlValue::Decimal(value)) => {
let value = value
.rescale(*scale)
.ok_or_else(|| StorageError::TypeMismatch {
expected: format!("Decimal({precision},{scale})"),
actual: "Decimal overflow".into(),
})?;
if value.fits_precision(*precision) {
Ok(SqlValue::Decimal(value))
} else {
Err(StorageError::TypeMismatch {
expected: format!("Decimal({precision},{scale})"),
actual: value.to_string(),
})
}
}
(Json, SqlValue::Json(value)) => Ok(SqlValue::Json(value)),
(Array(element), SqlValue::Array(values)) => values
.into_iter()
.map(|value| ensure_type(value, element))
.collect::<Result<Vec<_>>>()
.map(SqlValue::Array),
(
Map {
key: key_type,
value: value_type,
},
SqlValue::Map(values),
) => values
.into_iter()
.map(|(key, value)| {
if key.is_null() {
return Err(StorageError::TypeMismatch {
expected: "non-NULL map key".into(),
actual: "Null".into(),
});
}
Ok((ensure_type(key, key_type)?, ensure_type(value, value_type)?))
})
.collect::<Result<Vec<_>>>()
.map(SqlValue::Map),
(Struct(fields), SqlValue::Struct(values)) if fields.len() == values.len() => values
.into_iter()
.zip(fields)
.map(|((name, value), (expected_name, expected_type))| {
if name != *expected_name {
return Err(StorageError::TypeMismatch {
expected: expected_name.clone(),
actual: name,
});
}
Ok((name, ensure_type(value, expected_type)?))
})
.collect::<Result<Vec<_>>>()
.map(SqlValue::Struct),
(Vector { dimension, .. }, SqlValue::Vector(values)) => {
if values.len() as u32 == *dimension {
Ok(SqlValue::Vector(values))
} else {
Err(StorageError::TypeMismatch {
expected: format!("Vector(dim={})", dimension),
actual: format!("Vector(dim={})", values.len()),
})
}
}
(expected_ty, actual) => Err(StorageError::TypeMismatch {
expected: expected_ty.type_name().to_string(),
actual: actual.type_name().to_string(),
}),
}
}
#[cfg(test)]
mod tests {
use super::*;
use proptest::prelude::*;
fn values_equal(a: &SqlValue, b: &SqlValue) -> bool {
match (a, b) {
(SqlValue::Float(x), SqlValue::Float(y)) => x.to_bits() == y.to_bits(),
(SqlValue::Double(x), SqlValue::Double(y)) => x.to_bits() == y.to_bits(),
(SqlValue::Vector(xs), SqlValue::Vector(ys)) => {
xs.len() == ys.len()
&& xs
.iter()
.zip(ys.iter())
.all(|(x, y)| x.to_bits() == y.to_bits())
}
_ => a == b,
}
}
fn row_equal(a: &[SqlValue], b: &[SqlValue]) -> bool {
a.len() == b.len()
&& a.iter()
.zip(b.iter())
.all(|(lhs, rhs)| values_equal(lhs, rhs))
}
fn sql_value_strategy() -> impl Strategy<Value = SqlValue> {
let finite_f32 = any::<f32>();
let finite_f64 = any::<f64>();
let decimal_max = crate::storage::value::decimal_power(38).unwrap() - 1;
prop_oneof![
Just(SqlValue::Null),
any::<i32>().prop_map(SqlValue::Integer),
any::<i64>().prop_map(SqlValue::BigInt),
finite_f32.prop_map(SqlValue::Float),
finite_f64.prop_map(SqlValue::Double),
".*".prop_map(SqlValue::Text),
proptest::collection::vec(any::<u8>(), 0..32).prop_map(SqlValue::Blob),
any::<bool>().prop_map(SqlValue::Boolean),
any::<i64>().prop_map(SqlValue::Timestamp),
proptest::collection::vec(any::<f32>(), 0..8).prop_map(SqlValue::Vector),
any::<i32>().prop_map(SqlValue::Date),
any::<i64>().prop_map(SqlValue::Time),
(any::<i32>(), any::<i32>(), any::<i64>()).prop_map(|(months, days, micros)| {
SqlValue::Interval {
months,
days,
micros,
}
}),
(-decimal_max..=decimal_max, 0_u8..=38).prop_map(|(coefficient, scale)| {
SqlValue::Decimal(crate::storage::DecimalValue::new(coefficient, scale))
}),
]
}
#[test]
fn roundtrip_preserves_all_types() {
let row = vec![
SqlValue::Null,
SqlValue::Integer(42),
SqlValue::BigInt(-42),
SqlValue::Float(1.5),
SqlValue::Double(-2.5),
SqlValue::Text("hello".into()),
SqlValue::Blob(vec![0x01, 0x02]),
SqlValue::Boolean(true),
SqlValue::Timestamp(1_700_000_000),
SqlValue::Vector(vec![0.1, 0.2, 0.3]),
SqlValue::Date(19_782),
SqlValue::Time(86_399_123_456),
SqlValue::Interval {
months: -1,
days: 2,
micros: 3,
},
SqlValue::Decimal(crate::storage::DecimalValue::new(-12345, 2)),
SqlValue::Array(vec![SqlValue::Integer(1), SqlValue::Null]),
SqlValue::Map(vec![(SqlValue::Text("a".into()), SqlValue::Integer(1))]),
SqlValue::Struct(vec![(
"items".into(),
SqlValue::Array(vec![SqlValue::Text("x".into())]),
)]),
];
let encoded = RowCodec::encode(&row);
let decoded = RowCodec::decode(&encoded).unwrap();
assert!(row_equal(&row, &decoded));
}
#[test]
fn pre_temporal_row_bytes_remain_readable() {
let bytes = [
2, 0, 0, 0x01, 42, 0, 0, 0, 0x08, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, ];
assert_eq!(
RowCodec::decode(&bytes).unwrap(),
vec![SqlValue::Integer(42), SqlValue::Timestamp(-1)]
);
}
#[test]
fn null_bitmap_is_respected() {
let row = vec![SqlValue::Integer(1), SqlValue::Null, SqlValue::Integer(2)];
let encoded = RowCodec::encode(&row);
let decoded = RowCodec::decode(&encoded).unwrap();
assert!(matches!(decoded[1], SqlValue::Null));
}
#[test]
fn corruption_is_detected_for_truncated_payload() {
let row = vec![SqlValue::Text("abc".into())];
let mut encoded = RowCodec::encode(&row);
encoded.pop(); let err = RowCodec::decode(&encoded).unwrap_err();
assert!(matches!(err, StorageError::CorruptedData { .. }));
}
#[test]
fn corruption_is_detected_for_unknown_tag() {
let bytes = vec![1, 0, 0, 0xFF];
let err = RowCodec::decode(&bytes).unwrap_err();
assert!(matches!(err, StorageError::CorruptedData { .. }));
}
#[test]
fn oversized_lengths_are_rejected() {
let mut bytes = Vec::new();
bytes.extend_from_slice(&(1u16).to_le_bytes()); bytes.push(0); bytes.push(0x05); let too_large = (super::MAX_INLINE_BYTES as u32) + 1;
bytes.extend_from_slice(&too_large.to_le_bytes());
let err = RowCodec::decode(&bytes).unwrap_err();
assert!(matches!(err, StorageError::CorruptedData { .. }));
}
#[test]
fn oversized_vector_is_rejected() {
let mut bytes = Vec::new();
bytes.extend_from_slice(&(1u16).to_le_bytes()); bytes.push(0); bytes.push(0x09); let too_large = (super::MAX_VECTOR_LEN as u32) + 1;
bytes.extend_from_slice(&too_large.to_le_bytes());
let err = RowCodec::decode(&bytes).unwrap_err();
assert!(matches!(err, StorageError::CorruptedData { .. }));
}
#[test]
fn decode_with_schema_validates_types() {
let row = vec![SqlValue::Vector(vec![1.0, 2.0])];
let encoded = RowCodec::encode(&row);
let schema = vec![ResolvedType::Vector {
dimension: 3,
metric: crate::ast::ddl::VectorMetric::Cosine,
}];
let err = RowCodec::decode_with_schema(&encoded, &schema).unwrap_err();
assert!(matches!(err, StorageError::TypeMismatch { .. }));
}
proptest! {
#[test]
fn proptest_roundtrip(row in proptest::collection::vec(sql_value_strategy(), 0..16)) {
let encoded = RowCodec::encode(&row);
let decoded = RowCodec::decode(&encoded).unwrap();
prop_assert!(row_equal(&row, &decoded));
}
#[test]
fn decode_with_schema_matches_lengths(row in proptest::collection::vec(sql_value_strategy(), 1..5)) {
let schema: Vec<ResolvedType> = row.iter().map(|v| v.resolved_type()).collect();
let encoded = RowCodec::encode(&row);
let decoded = RowCodec::decode_with_schema(&encoded, &schema).unwrap();
prop_assert!(row_equal(&row, &decoded));
}
}
}