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