1use std::{collections::HashMap, str::from_utf8};
5
6use arrow_buffer::i256;
7use reifydb_value::{
8 encoding::LeBytes,
9 params::Params,
10 value::{
11 Value,
12 blob::Blob,
13 date::Date,
14 datetime::DateTime,
15 decimal::Decimal,
16 dictionary::DictionaryEntryId,
17 digest::Digest,
18 duration::Duration,
19 enum_variant::EnumVariant,
20 identity::IdentityId,
21 ordered_f32::OrderedF32,
22 ordered_f64::OrderedF64,
23 time::Time,
24 uuid::{Uuid4, Uuid7},
25 },
26};
27
28use crate::{
29 error::{DecodeError, EncodeError},
30 reader::Reader,
31 tag::ValueKind,
32 typeinfo::{decode_value_type, encode_value_type},
33};
34
35pub fn encode_value(value: &Value) -> Result<Vec<u8>, EncodeError> {
36 let mut buf = Vec::new();
37 encode_value_into(value, &mut buf)?;
38 Ok(buf)
39}
40
41pub fn encode_value_into(value: &Value, buf: &mut Vec<u8>) -> Result<(), EncodeError> {
42 buf.push(ValueKind::of_value(value).byte());
43 match value {
44 Value::None {
45 inner,
46 } => encode_value_type(inner, buf)?,
47 Value::Boolean(b) => buf.push(*b as u8),
48 Value::Float4(f) => buf.extend_from_slice(&f.to_le_bytes()),
49 Value::Float8(f) => buf.extend_from_slice(&f.to_le_bytes()),
50 Value::Int1(v) => buf.extend_from_slice(&v.to_le_bytes()),
51 Value::Int2(v) => buf.extend_from_slice(&v.to_le_bytes()),
52 Value::Int4(v) => buf.extend_from_slice(&v.to_le_bytes()),
53 Value::Int8(v) => buf.extend_from_slice(&v.to_le_bytes()),
54 Value::Int16(v) => buf.extend_from_slice(&v.to_le_bytes()),
55 Value::Utf8(s) => encode_len_prefixed(s.as_bytes(), buf),
56 Value::Enum(value) => {
57 encode_len_prefixed(value.r#enum.as_bytes(), buf);
58 encode_len_prefixed(value.variant.as_bytes(), buf);
59 }
60 Value::Uint1(v) => buf.extend_from_slice(&v.to_le_bytes()),
61 Value::Uint2(v) => buf.extend_from_slice(&v.to_le_bytes()),
62 Value::Uint4(v) => buf.extend_from_slice(&v.to_le_bytes()),
63 Value::Uint8(v) => buf.extend_from_slice(&v.to_le_bytes()),
64 Value::Uint16(v) => buf.extend_from_slice(&v.to_le_bytes()),
65 Value::Date(d) => buf.extend_from_slice(d.to_le_bytes().as_ref()),
66 Value::DateTime(dt) => buf.extend_from_slice(dt.to_le_bytes().as_ref()),
67 Value::Time(t) => buf.extend_from_slice(t.to_le_bytes().as_ref()),
68 Value::Duration(d) => buf.extend_from_slice(d.to_le_bytes().as_ref()),
69 Value::IdentityId(id) => buf.extend_from_slice(id.to_le_bytes().as_ref()),
70 Value::Uuid4(u) => buf.extend_from_slice(u.to_le_bytes().as_ref()),
71 Value::Uuid7(u) => buf.extend_from_slice(u.to_le_bytes().as_ref()),
72 Value::Blob(b) => encode_len_prefixed(b.as_bytes(), buf),
73 Value::Decimal(v) => {
74 buf.push(v.scale());
75 buf.extend_from_slice(&v.unscaled().to_le_bytes());
76 }
77 Value::Any(inner) => encode_value_into(inner, buf)?,
78 Value::DictionaryId(id) => match id {
79 DictionaryEntryId::U1(v) => {
80 buf.push(1);
81 buf.extend_from_slice(&v.to_le_bytes());
82 }
83 DictionaryEntryId::U2(v) => {
84 buf.push(2);
85 buf.extend_from_slice(&v.to_le_bytes());
86 }
87 DictionaryEntryId::U4(v) => {
88 buf.push(4);
89 buf.extend_from_slice(&v.to_le_bytes());
90 }
91 DictionaryEntryId::U8(v) => {
92 buf.push(8);
93 buf.extend_from_slice(&v.to_le_bytes());
94 }
95 DictionaryEntryId::U16(v) => {
96 buf.push(16);
97 buf.extend_from_slice(&v.to_le_bytes());
98 }
99 },
100 Value::Type(ty) => encode_value_type(ty, buf)?,
101 Value::List(items) | Value::Tuple(items) => {
102 buf.extend_from_slice(&(items.len() as u32).to_le_bytes());
103 for item in items {
104 encode_value_into(item, buf)?;
105 }
106 }
107 Value::Record(fields) => {
108 buf.extend_from_slice(&(fields.len() as u32).to_le_bytes());
109 for (key, field_value) in fields {
110 encode_len_prefixed(key.as_bytes(), buf);
111 encode_value_into(field_value, buf)?;
112 }
113 }
114 Value::Digest(digest) => encode_len_prefixed(&digest.encode(), buf),
115 }
116 Ok(())
117}
118
119fn encode_len_prefixed(bytes: &[u8], buf: &mut Vec<u8>) {
120 buf.extend_from_slice(&(bytes.len() as u32).to_le_bytes());
121 buf.extend_from_slice(bytes);
122}
123
124pub fn decode_value(bytes: &[u8]) -> Result<Value, DecodeError> {
125 let mut r = Reader::new(bytes);
126 let value = decode_value_from(&mut r)?;
127 if !r.is_empty() {
128 return Err(DecodeError::TrailingBytes(r.remaining()));
129 }
130 Ok(value)
131}
132
133pub fn decode_value_from(r: &mut Reader) -> Result<Value, DecodeError> {
134 let byte = r.u8()?;
135 let kind = ValueKind::from_byte(byte).ok_or(DecodeError::UnknownTypeCode(byte))?;
136 match kind {
137 ValueKind::None => Ok(Value::None {
138 inner: decode_value_type(r)?,
139 }),
140 ValueKind::Boolean => Ok(Value::Boolean(r.u8()? != 0)),
141 ValueKind::Float4 => OrderedF32::try_from(r.f32()?)
142 .map(Value::Float4)
143 .map_err(|e| DecodeError::InvalidData(format!("invalid float4: {e}"))),
144 ValueKind::Float8 => OrderedF64::try_from(r.f64()?)
145 .map(Value::Float8)
146 .map_err(|e| DecodeError::InvalidData(format!("invalid float8: {e}"))),
147 ValueKind::Int1 => Ok(Value::Int1(r.i8()?)),
148 ValueKind::Int2 => Ok(Value::Int2(r.i16()?)),
149 ValueKind::Int4 => Ok(Value::Int4(r.i32()?)),
150 ValueKind::Int8 => Ok(Value::Int8(r.i64()?)),
151 ValueKind::Int16 => Ok(Value::Int16(r.i128()?)),
152 ValueKind::Utf8 => Ok(Value::Utf8(decode_len_prefixed_str(r)?.to_string())),
153 ValueKind::Uint1 => Ok(Value::Uint1(r.u8()?)),
154 ValueKind::Uint2 => Ok(Value::Uint2(r.u16()?)),
155 ValueKind::Uint4 => Ok(Value::Uint4(r.u32()?)),
156 ValueKind::Uint8 => Ok(Value::Uint8(r.u64()?)),
157 ValueKind::Uint16 => Ok(Value::Uint16(r.u128()?)),
158 ValueKind::Date => {
159 let days = r.i32()?;
160 Date::from_days_since_epoch(days)
161 .map(Value::Date)
162 .ok_or_else(|| DecodeError::InvalidData(format!("invalid date: {days}")))
163 }
164 ValueKind::DateTime => Ok(Value::DateTime(DateTime::read_le(r.take(DateTime::ENCODED_SIZE)?))),
165 ValueKind::Time => {
166 let nanos = r.u64()?;
167 Time::from_nanos_since_midnight(nanos)
168 .map(Value::Time)
169 .ok_or_else(|| DecodeError::InvalidData(format!("invalid time: {nanos}")))
170 }
171 ValueKind::Duration => {
172 let months = r.i32()?;
173 let days = r.i32()?;
174 let nanos = r.i64()?;
175 Duration::new(months, days, nanos)
176 .map(Value::Duration)
177 .map_err(|e| DecodeError::InvalidData(format!("invalid duration: {e}")))
178 }
179 ValueKind::IdentityId => Ok(Value::IdentityId(IdentityId::read_le(r.take(IdentityId::ENCODED_SIZE)?))),
180 ValueKind::Uuid4 => Ok(Value::Uuid4(Uuid4::read_le(r.take(Uuid4::ENCODED_SIZE)?))),
181 ValueKind::Uuid7 => Ok(Value::Uuid7(Uuid7::read_le(r.take(Uuid7::ENCODED_SIZE)?))),
182 ValueKind::Blob => Ok(Value::Blob(Blob::new(decode_len_prefixed_bytes(r)?.to_vec()))),
183 ValueKind::Decimal => {
184 let scale = r.u8()?;
185 let unscaled = decode_i256(r)?;
186 Decimal::from_parts(unscaled, scale).map(Value::Decimal).ok_or_else(|| {
187 DecodeError::InvalidData(format!(
188 "decimal {unscaled} at scale {scale} exceeds 76 digits"
189 ))
190 })
191 }
192 ValueKind::Any => Ok(Value::Any(Box::new(decode_value_from(r)?))),
193 ValueKind::DictionaryId => {
194 let width = r.u8()?;
195 let id = match width {
196 1 => DictionaryEntryId::U1(r.u8()?),
197 2 => DictionaryEntryId::U2(r.u16()?),
198 4 => DictionaryEntryId::U4(r.u32()?),
199 8 => DictionaryEntryId::U8(r.u64()?),
200 16 => DictionaryEntryId::U16(r.u128()?),
201 other => {
202 return Err(DecodeError::InvalidData(format!(
203 "invalid dictionary id width: {other}"
204 )));
205 }
206 };
207 Ok(Value::DictionaryId(id))
208 }
209 ValueKind::Type => Ok(Value::Type(decode_value_type(r)?)),
210 ValueKind::List => Ok(Value::List(decode_value_sequence(r)?)),
211 ValueKind::Tuple => Ok(Value::Tuple(decode_value_sequence(r)?)),
212 ValueKind::Record => {
213 let count = r.u32()?;
214 let mut fields = Vec::with_capacity((count as usize).min(4096));
215 for _ in 0..count {
216 let key = decode_len_prefixed_str(r)?.to_string();
217 fields.push((key, decode_value_from(r)?));
218 }
219 Ok(Value::Record(fields))
220 }
221 ValueKind::Enum => {
222 let r#enum = decode_len_prefixed_str(r)?.to_string();
223 Ok(Value::Enum(EnumVariant::new(r#enum, decode_len_prefixed_str(r)?)))
224 }
225 ValueKind::Digest => Digest::decode(decode_len_prefixed_bytes(r)?)
226 .map(|digest| Value::Digest(Box::new(digest)))
227 .map_err(|error| DecodeError::InvalidData(format!("invalid digest: {error}"))),
228 }
229}
230
231fn decode_i256(r: &mut Reader) -> Result<i256, DecodeError> {
232 let mut bytes = [0u8; 32];
233 bytes.copy_from_slice(r.take(32)?);
234 Ok(i256::from_le_bytes(bytes))
235}
236
237fn decode_value_sequence(r: &mut Reader) -> Result<Vec<Value>, DecodeError> {
238 let count = r.u32()?;
239 let mut items = Vec::with_capacity((count as usize).min(4096));
240 for _ in 0..count {
241 items.push(decode_value_from(r)?);
242 }
243 Ok(items)
244}
245
246fn decode_len_prefixed_bytes<'a>(r: &mut Reader<'a>) -> Result<&'a [u8], DecodeError> {
247 let len = r.u32()? as usize;
248 r.take(len)
249}
250
251fn decode_len_prefixed_str<'a>(r: &mut Reader<'a>) -> Result<&'a str, DecodeError> {
252 from_utf8(decode_len_prefixed_bytes(r)?).map_err(|e| DecodeError::InvalidData(format!("invalid UTF-8: {e}")))
253}
254
255pub fn encode_params(params: &Params) -> Result<Vec<u8>, EncodeError> {
256 let mut buf = Vec::new();
257 match params {
258 Params::None => buf.push(0),
259 Params::Positional(values) => {
260 buf.push(1);
261 buf.extend_from_slice(&(values.len() as u32).to_le_bytes());
262 for value in values.iter() {
263 encode_value_into(value, &mut buf)?;
264 }
265 }
266 Params::Named(map) => {
267 buf.push(2);
268 buf.extend_from_slice(&(map.len() as u32).to_le_bytes());
269 for (key, value) in map.iter() {
270 encode_len_prefixed(key.as_bytes(), &mut buf);
271 encode_value_into(value, &mut buf)?;
272 }
273 }
274 }
275 Ok(buf)
276}
277
278pub fn decode_params(bytes: &[u8]) -> Result<Params, DecodeError> {
279 let mut r = Reader::new(bytes);
280 let discriminant = r.u8()?;
281 let params = match discriminant {
282 0 => Params::None,
283 1 => {
284 let count = r.u32()?;
285 let mut values = Vec::with_capacity((count as usize).min(4096));
286 for _ in 0..count {
287 values.push(decode_value_from(&mut r)?);
288 }
289 Params::Positional(values.into())
290 }
291 2 => {
292 let count = r.u32()?;
293 let mut map = HashMap::with_capacity((count as usize).min(4096));
294 for _ in 0..count {
295 let key = decode_len_prefixed_str(&mut r)?.to_string();
296 map.insert(key, decode_value_from(&mut r)?);
297 }
298 Params::Named(map.into())
299 }
300 other => {
301 return Err(DecodeError::InvalidData(format!("invalid params discriminant: {other}")));
302 }
303 };
304 if !r.is_empty() {
305 return Err(DecodeError::TrailingBytes(r.remaining()));
306 }
307 Ok(params)
308}