1use bigdecimal::BigDecimal as BigDecimalInner;
5use num_bigint::{BigInt, Sign};
6use reifydb_value::{
7 Result,
8 error::{Error, TypeError},
9 value::{
10 Value,
11 blob::Blob,
12 date::Date,
13 datetime::DateTime,
14 decimal::Decimal,
15 dictionary::DictionaryEntryId,
16 duration::Duration,
17 identity::IdentityId,
18 int::Int,
19 ordered_f32::OrderedF32,
20 ordered_f64::OrderedF64,
21 row_number::RowNumber,
22 time::Time,
23 uint::Uint,
24 uuid::{Uuid4, Uuid7},
25 },
26};
27use uuid::Uuid;
28
29use super::{
30 CONTAINER_END, decode_bool, decode_f32, decode_f64, decode_fixed, decode_i8, decode_i16, decode_i32,
31 decode_i64, decode_i128, decode_u8, decode_u16, decode_u32, decode_u64, decode_u128, decode_u128_varint,
32};
33use crate::{
34 key::serializer::{DECIMAL_END_NEGATIVE, DECIMAL_END_POSITIVE},
35 tag::{TypeTag, ValueKind},
36};
37
38pub struct KeyDeserializer<'a> {
39 buffer: &'a [u8],
40 position: usize,
41}
42
43impl<'a> KeyDeserializer<'a> {
44 pub fn from_bytes(buffer: &'a [u8]) -> Self {
45 Self {
46 buffer,
47 position: 0,
48 }
49 }
50
51 pub fn remaining(&self) -> usize {
52 self.buffer.len().saturating_sub(self.position)
53 }
54
55 pub fn is_empty(&self) -> bool {
56 self.remaining() == 0
57 }
58
59 pub fn position(&self) -> usize {
60 self.position
61 }
62
63 pub fn remaining_bytes(&self) -> &'a [u8] {
64 &self.buffer[self.position..]
65 }
66
67 fn read_exact(&mut self, count: usize) -> Result<&'a [u8]> {
68 if self.remaining() < count {
69 return Err(Error::from(TypeError::SerdeKeycode {
70 message: format!(
71 "unexpected end of key at position {}: need {} bytes, have {}",
72 self.position,
73 count,
74 self.remaining()
75 ),
76 }));
77 }
78 let start = self.position;
79 self.position += count;
80 Ok(&self.buffer[start..self.position])
81 }
82
83 pub fn read_bool(&mut self) -> Result<bool> {
84 let bytes = self.read_exact(1)?;
85 decode_bool(bytes[0])
86 }
87
88 pub fn read_f32(&mut self) -> Result<f32> {
89 let bytes = self.read_exact(4)?;
90 Ok(decode_f32(bytes.try_into()?))
91 }
92
93 pub fn read_f64(&mut self) -> Result<f64> {
94 let bytes = self.read_exact(8)?;
95 Ok(decode_f64(bytes.try_into()?))
96 }
97
98 pub fn read_i8(&mut self) -> Result<i8> {
99 let bytes = self.read_exact(1)?;
100 Ok(decode_i8(bytes.try_into()?))
101 }
102
103 pub fn read_i16(&mut self) -> Result<i16> {
104 let bytes = self.read_exact(2)?;
105 Ok(decode_i16(bytes.try_into()?))
106 }
107
108 pub fn read_i32(&mut self) -> Result<i32> {
109 let bytes = self.read_exact(4)?;
110 Ok(decode_i32(bytes.try_into()?))
111 }
112
113 pub fn read_i64(&mut self) -> Result<i64> {
114 let bytes = self.read_exact(8)?;
115 Ok(decode_i64(bytes.try_into()?))
116 }
117
118 pub fn read_i128(&mut self) -> Result<i128> {
119 let bytes = self.read_exact(16)?;
120 Ok(decode_i128(bytes.try_into()?))
121 }
122
123 pub fn read_u8(&mut self) -> Result<u8> {
124 let bytes = self.read_exact(1)?;
125 Ok(decode_u8(bytes[0]))
126 }
127
128 pub fn read_u16(&mut self) -> Result<u16> {
129 let bytes = self.read_exact(2)?;
130 Ok(decode_u16(bytes.try_into()?))
131 }
132
133 pub fn read_u32(&mut self) -> Result<u32> {
134 let bytes = self.read_exact(4)?;
135 Ok(decode_u32(bytes.try_into()?))
136 }
137
138 pub fn read_u64(&mut self) -> Result<u64> {
139 let bytes = self.read_exact(8)?;
140 Ok(decode_u64(bytes.try_into()?))
141 }
142
143 pub fn read_u128(&mut self) -> Result<u128> {
144 let bytes = self.read_exact(16)?;
145 Ok(decode_u128(bytes.try_into()?))
146 }
147
148 pub fn read_fixed<const N: usize>(&mut self) -> Result<[u8; N]> {
149 let bytes = self.read_exact(N)?;
150 Ok(decode_fixed(bytes.try_into()?))
151 }
152
153 pub fn read_u128_varint(&mut self) -> Result<u128> {
154 let mut slice = &self.buffer[self.position..];
155 let u = decode_u128_varint(&mut slice)?;
156 self.position = self.buffer.len() - slice.len();
157 Ok(u)
158 }
159
160 pub fn read_bytes(&mut self) -> Result<Vec<u8>> {
161 let mut result = Vec::new();
162 loop {
163 if self.remaining() < 1 {
164 return Err(Error::from(TypeError::SerdeKeycode {
165 message: format!(
166 "unexpected end of key at position {}: bytes not terminated",
167 self.position
168 ),
169 }));
170 }
171 let byte = self.buffer[self.position];
172 self.position += 1;
173
174 if byte == 0xff {
175 if self.remaining() < 1 {
176 return Err(Error::from(TypeError::SerdeKeycode {
177 message: format!(
178 "unexpected end of key at position {}: incomplete escape sequence",
179 self.position
180 ),
181 }));
182 }
183 let next_byte = self.buffer[self.position];
184 self.position += 1;
185
186 if next_byte == 0x00 {
187 result.push(0x00);
188 } else if next_byte == 0xff {
189 break;
190 } else {
191 return Err(Error::from(TypeError::SerdeKeycode {
192 message: format!(
193 "invalid escape sequence at position {}: 0xff 0x{:02x}",
194 self.position - 1,
195 next_byte
196 ),
197 }));
198 }
199 } else {
200 result.push(!byte);
201 }
202 }
203 Ok(result)
204 }
205
206 pub fn read_str(&mut self) -> Result<String> {
207 let bytes = self.read_bytes()?;
208 String::from_utf8(bytes).map_err(|e| {
209 Error::from(TypeError::SerdeKeycode {
210 message: format!("invalid UTF-8 in key at position {}: {}", self.position, e),
211 })
212 })
213 }
214
215 pub fn read_date(&mut self) -> Result<Date> {
216 let days = self.read_i32()?;
217 Date::from_days_since_epoch(days).ok_or_else(|| {
218 Error::from(TypeError::SerdeKeycode {
219 message: format!(
220 "invalid date at position {}: {} days since epoch",
221 self.position, days
222 ),
223 })
224 })
225 }
226
227 pub fn read_datetime(&mut self) -> Result<DateTime> {
228 let nanos = self.read_u64()?;
229 Ok(DateTime::from_nanos(nanos))
230 }
231
232 pub fn read_time(&mut self) -> Result<Time> {
233 let nanos = self.read_u64()?;
234 Time::from_nanos_since_midnight(nanos).ok_or_else(|| {
235 Error::from(TypeError::SerdeKeycode {
236 message: format!(
237 "invalid time at position {}: {} nanos since midnight",
238 self.position, nanos
239 ),
240 })
241 })
242 }
243
244 pub fn read_duration(&mut self) -> Result<Duration> {
245 let months = self.read_i32()?;
246 let days = self.read_i32()?;
247 let nanos = self.read_i64()?;
248 Ok(Duration::new(months, days, nanos)?)
249 }
250
251 pub fn read_row_number(&mut self) -> Result<RowNumber> {
252 let value = self.read_u64()?;
253 Ok(RowNumber(value))
254 }
255
256 pub fn read_identity_id(&mut self) -> Result<IdentityId> {
257 let bytes = self.read_bytes()?;
258 let uuid = Uuid::from_slice(&bytes).map_err(|e| {
259 Error::from(TypeError::SerdeKeycode {
260 message: format!("invalid IdentityId at position {}: {}", self.position, e),
261 })
262 })?;
263 Ok(IdentityId::from(Uuid7::from(uuid)))
264 }
265
266 pub fn read_uuid4(&mut self) -> Result<Uuid4> {
267 let bytes = self.read_bytes()?;
268 let uuid = Uuid::from_slice(&bytes).map_err(|e| {
269 Error::from(TypeError::SerdeKeycode {
270 message: format!("invalid Uuid4 at position {}: {}", self.position, e),
271 })
272 })?;
273 Ok(Uuid4::from(uuid))
274 }
275
276 pub fn read_uuid7(&mut self) -> Result<Uuid7> {
277 let bytes = self.read_bytes()?;
278 let uuid = Uuid::from_slice(&bytes).map_err(|e| {
279 Error::from(TypeError::SerdeKeycode {
280 message: format!("invalid Uuid7 at position {}: {}", self.position, e),
281 })
282 })?;
283 Ok(Uuid7::from(uuid))
284 }
285
286 pub fn read_blob(&mut self) -> Result<Blob> {
287 let bytes = self.read_bytes()?;
288 Ok(Blob::from(bytes))
289 }
290
291 pub fn read_int(&mut self) -> Result<Int> {
292 if decode_u8(self.read_exact(1)?[0]) == 0 {
293 let len = u32::from_be_bytes(self.read_exact(4)?.try_into()?) as usize;
294 let bytes = self.read_exact(len)?;
295 return Ok(Int(BigInt::from_bytes_be(Sign::Minus, bytes)));
296 }
297 let len = self.read_u32()? as usize;
298 let bytes: Vec<u8> = self.read_exact(len)?.iter().map(|byte| decode_u8(*byte)).collect();
299 Ok(Int(BigInt::from_bytes_be(Sign::Plus, &bytes)))
300 }
301
302 pub fn read_uint(&mut self) -> Result<Uint> {
303 let len = self.read_u32()? as usize;
304 let bytes: Vec<u8> = self.read_exact(len)?.iter().map(|byte| decode_u8(*byte)).collect();
305 Ok(Uint(BigInt::from_bytes_be(Sign::Plus, &bytes)))
306 }
307
308 pub fn read_decimal(&mut self) -> Result<Decimal> {
309 let negative = decode_u8(self.read_exact(1)?[0]) == 0;
310 self.read_exact(4)?;
311
312 let terminator = if negative {
313 DECIMAL_END_NEGATIVE
314 } else {
315 DECIMAL_END_POSITIVE
316 };
317 let mut digits = Vec::new();
318 loop {
319 let byte = self.read_exact(1)?[0];
320 if byte == terminator {
321 break;
322 }
323 digits.push(if negative {
324 byte
325 } else {
326 decode_u8(byte)
327 });
328 }
329 let scale = self.read_i64()?;
330
331 let mantissa = if digits.is_empty() {
332 BigInt::from(0)
333 } else {
334 let magnitude = BigInt::parse_bytes(&digits, 10).ok_or_else(|| {
335 Error::from(TypeError::SerdeKeycode {
336 message: format!("invalid Decimal digits at position {}", self.position),
337 })
338 })?;
339 if negative {
340 -magnitude
341 } else {
342 magnitude
343 }
344 };
345 Ok(Decimal(BigDecimalInner::new(mantissa, scale)))
346 }
347
348 fn at_container_end(&mut self) -> Result<bool> {
349 if self.remaining() < 1 {
350 return Err(Error::from(TypeError::SerdeKeycode {
351 message: format!(
352 "unexpected end of key at position {}: container not terminated",
353 self.position
354 ),
355 }));
356 }
357 if self.buffer[self.position] == CONTAINER_END {
358 self.position += 1;
359 return Ok(true);
360 }
361 Ok(false)
362 }
363
364 fn read_container_items(&mut self) -> Result<Vec<Value>> {
365 let mut items = Vec::new();
366 while !self.at_container_end()? {
367 items.push(self.read_value()?);
368 }
369 Ok(items)
370 }
371
372 fn read_record_fields(&mut self) -> Result<Vec<(String, Value)>> {
373 let mut fields = Vec::new();
374 while !self.at_container_end()? {
375 let name = self.read_str()?;
376 fields.push((name, self.read_value()?));
377 }
378 Ok(fields)
379 }
380
381 pub fn read_value(&mut self) -> Result<Value> {
382 if self.remaining() < 1 {
383 return Err(Error::from(TypeError::SerdeKeycode {
384 message: format!(
385 "unexpected end of key at position {}: cannot read value type",
386 self.position
387 ),
388 }));
389 }
390
391 let type_marker = self.buffer[self.position];
392 self.position += 1;
393
394 let kind = ValueKind::from_byte(type_marker).ok_or_else(|| {
395 Error::from(TypeError::SerdeKeycode {
396 message: format!(
397 "unknown value type marker 0x{:02x} at position {}",
398 type_marker,
399 self.position - 1
400 ),
401 })
402 })?;
403
404 match kind {
405 ValueKind::None => {
406 if self.remaining() < 1 {
407 return Ok(Value::none());
408 }
409 let inner_marker = self.buffer[self.position];
410 self.position += 1;
411 let inner = TypeTag::from_byte(inner_marker)
412 .map_err(|e| {
413 Error::from(TypeError::SerdeKeycode {
414 message: format!(
415 "invalid none inner type byte 0x{:02x} at position {}: {}",
416 inner_marker,
417 self.position - 1,
418 e
419 ),
420 })
421 })?
422 .to_type()
423 .map_err(|e| {
424 Error::from(TypeError::SerdeKeycode {
425 message: format!(
426 "invalid none inner type byte 0x{:02x} at position {}: {}",
427 inner_marker,
428 self.position - 1,
429 e
430 ),
431 })
432 })?;
433 Ok(Value::none_of(inner))
434 }
435 ValueKind::Float4 => {
436 let f = self.read_f32()?;
437 Ok(Value::Float4(OrderedF32::try_from(f).map_err(|e| {
438 Error::from(TypeError::SerdeKeycode {
439 message: format!("invalid f32 at position {}: {}", self.position, e),
440 })
441 })?))
442 }
443 ValueKind::Float8 => {
444 let f = self.read_f64()?;
445 Ok(Value::Float8(OrderedF64::try_from(f).map_err(|e| {
446 Error::from(TypeError::SerdeKeycode {
447 message: format!("invalid f64 at position {}: {}", self.position, e),
448 })
449 })?))
450 }
451 ValueKind::Boolean => Ok(Value::Boolean(self.read_bool()?)),
452 ValueKind::Int1 => Ok(Value::Int1(self.read_i8()?)),
453 ValueKind::Int2 => Ok(Value::Int2(self.read_i16()?)),
454 ValueKind::Int4 => Ok(Value::Int4(self.read_i32()?)),
455 ValueKind::Int8 => Ok(Value::Int8(self.read_i64()?)),
456 ValueKind::Int16 => Ok(Value::Int16(self.read_i128()?)),
457 ValueKind::Utf8 => Ok(Value::Utf8(self.read_str()?)),
458 ValueKind::Uint1 => Ok(Value::Uint1(self.read_u8()?)),
459 ValueKind::Uint2 => Ok(Value::Uint2(self.read_u16()?)),
460 ValueKind::Uint4 => Ok(Value::Uint4(self.read_u32()?)),
461 ValueKind::Uint8 => Ok(Value::Uint8(self.read_u64()?)),
462 ValueKind::Uint16 => Ok(Value::Uint16(self.read_u128()?)),
463 ValueKind::Date => Ok(Value::Date(self.read_date()?)),
464 ValueKind::DateTime => Ok(Value::DateTime(self.read_datetime()?)),
465 ValueKind::Time => Ok(Value::Time(self.read_time()?)),
466 ValueKind::Duration => Ok(Value::Duration(self.read_duration()?)),
467 ValueKind::IdentityId => Ok(Value::IdentityId(self.read_identity_id()?)),
468 ValueKind::Uuid4 => Ok(Value::Uuid4(self.read_uuid4()?)),
469 ValueKind::Uuid7 => Ok(Value::Uuid7(self.read_uuid7()?)),
470 ValueKind::Blob => Ok(Value::Blob(self.read_blob()?)),
471 ValueKind::Int => Ok(Value::Int(self.read_int()?)),
472 ValueKind::Uint => Ok(Value::Uint(self.read_uint()?)),
473 ValueKind::Decimal => Ok(Value::Decimal(self.read_decimal()?)),
474 ValueKind::List => Ok(Value::List(self.read_container_items()?)),
475 ValueKind::Tuple => Ok(Value::Tuple(self.read_container_items()?)),
476 ValueKind::Record => Ok(Value::Record(self.read_record_fields()?)),
477 ValueKind::Any | ValueKind::Type => Err(Error::from(TypeError::SerdeKeycode {
478 message: format!(
479 "value kind {:?} cannot be deserialized from keys (position {})",
480 kind,
481 self.position - 1
482 ),
483 })),
484 ValueKind::DictionaryId => {
485 let sub = self.read_exact(1)?[0];
486 match sub {
487 0x00 => Ok(Value::DictionaryId(DictionaryEntryId::U1(self.read_u8()?))),
488 0x01 => Ok(Value::DictionaryId(DictionaryEntryId::U2(self.read_u16()?))),
489 0x02 => Ok(Value::DictionaryId(DictionaryEntryId::U4(self.read_u32()?))),
490 0x03 => Ok(Value::DictionaryId(DictionaryEntryId::U8(self.read_u64()?))),
491 0x04 => Ok(Value::DictionaryId(DictionaryEntryId::U16(self.read_u128()?))),
492 _ => Err(Error::from(TypeError::SerdeKeycode {
493 message: format!(
494 "unknown DictionaryEntryId sub-marker 0x{:02x} at position {}",
495 sub,
496 self.position - 1
497 ),
498 })),
499 }
500 }
501 }
502 }
503
504 pub fn read_raw(&mut self, count: usize) -> Result<&'a [u8]> {
505 self.read_exact(count)
506 }
507}