reifydb_codec/json/
from.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::sync::Arc;
5
6use arrow_array::{
7	ArrayRef, BooleanArray, Float32Array, Float64Array, Int8Array, Int16Array, Int32Array, Int64Array,
8	LargeStringArray, UInt8Array, UInt16Array, UInt32Array, UInt64Array, builder::LargeBinaryBuilder,
9};
10use arrow_buffer::BooleanBuffer;
11use arrow_schema::FieldRef;
12use reifydb_value::{
13	fragment::Fragment,
14	util::hex::decode,
15	value::{
16		Value,
17		blob::Blob,
18		constraint::{precision::Precision, scale::Scale},
19		container::{
20			any_array::any_array_optional,
21			decimal_array::decimal_array,
22			digest_array::push_digest,
23			temporal_array::{date_array, datetime_array, duration_array, time_array},
24			uuid_array::{identity_id_array, uuid4_array, uuid7_array},
25			varlen_array::blob_array,
26			wide_int_array::wide_array,
27		},
28		date::Date,
29		datetime::DateTime,
30		decimal::{Decimal, parse::parse_decimal},
31		diff_type::DiffType,
32		digest::Digest,
33		duration::Duration,
34		enum_variant::EnumVariant,
35		frame::frame::Frame,
36		identity::IdentityId,
37		ordered_f32::OrderedF32,
38		ordered_f64::OrderedF64,
39		temporal::parse::{
40			date::parse_date, datetime::parse_datetime, duration::parse_duration, time::parse_time,
41		},
42		time::Time,
43		uuid::{
44			Uuid4, Uuid7,
45			parse::{parse_uuid4, parse_uuid7},
46		},
47		value_type::ValueType,
48	},
49};
50use serde::de::Error as _;
51use serde_json::{Error, Value as JsonValue, from_str, from_value};
52
53use crate::{
54	error::DecodeError,
55	frame::decode::{decoded_column, frame_batch},
56	json::{excerpt, none_marker_depth, types::ResponseFrame},
57	tag::peel_options,
58	unscaled::decimal_unscaled,
59};
60
61pub fn frames_from_json(json: &str) -> Result<Vec<Frame>, Error> {
62	let response_frames: Vec<ResponseFrame> = from_str(json)?;
63	frames_from_response(response_frames)
64}
65
66pub fn frames_from_envelope(body: JsonValue) -> Result<Vec<Frame>, Error> {
67	let mut envelope = match body {
68		JsonValue::Object(envelope) => envelope,
69		other => {
70			return Err(Error::custom(format!(
71				"frames envelope is not an object: {}",
72				excerpt(&other.to_string())
73			)));
74		}
75	};
76	let frames = envelope.remove("frames").ok_or_else(|| Error::custom("frames envelope has no `frames` key"))?;
77	frames_from_response(from_value(frames)?)
78}
79
80fn frames_from_response(response_frames: Vec<ResponseFrame>) -> Result<Vec<Frame>, Error> {
81	response_frames.into_iter().map(response_frame_to_frame).collect::<Result<_, _>>().map_err(Error::custom)
82}
83
84fn response_frame_to_frame(frame: ResponseFrame) -> Result<Frame, DecodeError> {
85	for (key, present) in [
86		("row_numbers", !frame.row_numbers.is_empty()),
87		("created_at", !frame.created_at.is_empty()),
88		("updated_at", !frame.updated_at.is_empty()),
89		("time", !frame.time.is_empty()),
90	] {
91		if present {
92			return Err(DecodeError::InvalidData(format!(
93				"frame key {key} is not supported, system columns travel as # columns"
94			)));
95		}
96	}
97
98	let columns = frame
99		.columns
100		.into_iter()
101		.map(|col| convert_column_to_data(&col.name, col.r#type.value_type, col.payload))
102		.collect::<Result<Vec<_>, DecodeError>>()?;
103	let row_count = columns.first().map_or(0, |(_, array)| array.len());
104
105	let op =
106		frame.op.map(|raw| {
107			DiffType::from_u8(raw)
108				.ok_or_else(|| DecodeError::InvalidData(format!("unknown frame op {raw}")))
109		})
110		.transpose()?;
111
112	Ok(Frame {
113		batch: frame_batch(columns, row_count)?,
114		op,
115	})
116}
117
118fn cell_error(column: &str, row: usize, ty: &ValueType, text: &str) -> DecodeError {
119	column_error(column, row, format!("cannot parse '{}' as {ty}", excerpt(text)))
120}
121
122fn column_error(column: &str, row: usize, reason: String) -> DecodeError {
123	DecodeError::ColumnDecodeFailed {
124		column_name: column.to_string(),
125		row_index: Some(row),
126		source: Box::new(DecodeError::InvalidData(reason)),
127	}
128}
129
130pub fn parse_value(ty: &ValueType, text: &str) -> Result<Value, DecodeError> {
131	let (base, depth) = peel_options(ty);
132	match none_marker_depth(text) {
133		Some(_) if depth == 0 => Err(DecodeError::InvalidData(format!("none marker for non-Option type {ty}"))),
134		Some(wrapped) if wrapped >= depth => Err(DecodeError::InvalidData(format!(
135			"none marker depth {wrapped} exceeds the {depth} Option layers of type {ty}"
136		))),
137		Some(wrapped) => Ok(Value::None {
138			inner: strip_options(ty, wrapped + 1),
139		}),
140		None => parse_base(base, text)
141			.ok_or_else(|| DecodeError::InvalidData(format!("cannot parse '{text}' as {ty}"))),
142	}
143}
144
145pub fn parse_json_value(ty: &ValueType, json: &JsonValue) -> Result<Value, DecodeError> {
146	let (base, depth) = peel_options(ty);
147	if let Some(wrapped) = json.as_str().and_then(none_marker_depth) {
148		return match wrapped {
149			_ if depth == 0 => {
150				Err(DecodeError::InvalidData(format!("none marker for non-Option type {ty}")))
151			}
152			wrapped if wrapped >= depth => Err(DecodeError::InvalidData(format!(
153				"none marker depth {wrapped} exceeds the {depth} Option layers of type {ty}"
154			))),
155			wrapped => Ok(Value::None {
156				inner: strip_options(ty, wrapped + 1),
157			}),
158		};
159	}
160	match base {
161		ValueType::List(inner) => {
162			let items = json
163				.as_array()
164				.ok_or_else(|| DecodeError::InvalidData(format!("cannot parse '{json}' as {ty}")))?;
165			let values = items
166				.iter()
167				.map(|item| parse_json_value(inner, item))
168				.collect::<Result<Vec<_>, _>>()?;
169			Ok(Value::List(values))
170		}
171		ValueType::Record(fields) => {
172			let obj = json
173				.as_object()
174				.ok_or_else(|| DecodeError::InvalidData(format!("cannot parse '{json}' as {ty}")))?;
175			let values = fields
176				.iter()
177				.map(|(name, field_ty)| {
178					let item = obj.get(name).ok_or_else(|| {
179						DecodeError::InvalidData(format!(
180							"record is missing field '{name}' for {ty}"
181						))
182					})?;
183					Ok((name.clone(), parse_json_value(field_ty, item)?))
184				})
185				.collect::<Result<Vec<_>, DecodeError>>()?;
186			Ok(Value::Record(values))
187		}
188		_ => {
189			let text = json.as_str().ok_or_else(|| {
190				DecodeError::InvalidData(format!("expected a JSON string for {ty}, got {json}"))
191			})?;
192			parse_value(ty, text)
193		}
194	}
195}
196
197fn strip_options(ty: &ValueType, layers: u32) -> ValueType {
198	let mut inner = ty;
199	for _ in 0..layers {
200		if let ValueType::Option(next) = inner {
201			inner = next;
202		}
203	}
204	inner.clone()
205}
206
207fn parse_base(base: &ValueType, text: &str) -> Option<Value> {
208	match base {
209		ValueType::Boolean => text.parse().ok().map(Value::Boolean),
210		ValueType::Float4 => parse_float4(text).map(Value::Float4),
211		ValueType::Float8 => parse_float8(text).map(Value::Float8),
212		ValueType::Int1 => text.parse().ok().map(Value::Int1),
213		ValueType::Int2 => text.parse().ok().map(Value::Int2),
214		ValueType::Int4 => text.parse().ok().map(Value::Int4),
215		ValueType::Int8 => text.parse().ok().map(Value::Int8),
216		ValueType::Int16 => text.parse().ok().map(Value::Int16),
217		ValueType::Utf8 => Some(Value::Utf8(text.to_string())),
218		ValueType::Enum(name) => Some(Value::Enum(EnumVariant::new(name.clone(), text))),
219		ValueType::Uint1 => text.parse().ok().map(Value::Uint1),
220		ValueType::Uint2 => text.parse().ok().map(Value::Uint2),
221		ValueType::Uint4 => text.parse().ok().map(Value::Uint4),
222		ValueType::Uint8 => text.parse().ok().map(Value::Uint8),
223		ValueType::Uint16 => text.parse().ok().map(Value::Uint16),
224		ValueType::Date => parse_date_text(text).map(Value::Date),
225		ValueType::DateTime => parse_datetime_text(text).map(Value::DateTime),
226		ValueType::Time => parse_time_text(text).map(Value::Time),
227		ValueType::Duration => parse_duration_text(text).map(Value::Duration),
228		ValueType::Uuid4 => parse_uuid4_text(text).map(Value::Uuid4),
229		ValueType::Uuid7 => parse_uuid7_text(text).map(Value::Uuid7),
230		ValueType::IdentityId => parse_identity_id(text).map(Value::IdentityId),
231		ValueType::Blob => parse_blob(text).map(Value::Blob),
232		ValueType::Decimal {
233			precision,
234			scale,
235		} => parse_decimal_text(text, *precision, *scale).map(Value::Decimal),
236		ValueType::Option(inner) => parse_base(inner, text),
237		ValueType::Digest {
238			inner,
239			accuracy,
240		} => parse_digest_text(inner, *accuracy, text).map(|digest| Value::Digest(Box::new(digest))),
241		ValueType::Any
242		| ValueType::DictionaryId
243		| ValueType::List(_)
244		| ValueType::Record(_)
245		| ValueType::Tuple(_) => Some(Value::Utf8(text.to_string())),
246	}
247}
248
249fn parse_float4(text: &str) -> Option<OrderedF32> {
250	OrderedF32::try_from(text.parse::<f32>().ok()?).ok()
251}
252
253fn parse_float8(text: &str) -> Option<OrderedF64> {
254	OrderedF64::try_from(text.parse::<f64>().ok()?).ok()
255}
256
257fn parse_date_text(text: &str) -> Option<Date> {
258	parse_date(Fragment::internal(text)).ok()
259}
260
261fn parse_datetime_text(text: &str) -> Option<DateTime> {
262	parse_datetime(Fragment::internal(text))
263		.ok()
264		.or_else(|| DateTime::from_epoch_secs(text.parse::<i64>().ok()?).ok())
265}
266
267fn parse_time_text(text: &str) -> Option<Time> {
268	parse_time(Fragment::internal(text)).ok()
269}
270
271fn parse_duration_text(text: &str) -> Option<Duration> {
272	parse_duration(Fragment::internal(text)).ok()
273}
274
275fn parse_uuid4_text(text: &str) -> Option<Uuid4> {
276	parse_uuid4(Fragment::internal(text)).ok()
277}
278
279fn parse_uuid7_text(text: &str) -> Option<Uuid7> {
280	parse_uuid7(Fragment::internal(text)).ok()
281}
282
283fn parse_identity_id(text: &str) -> Option<IdentityId> {
284	parse_uuid7_text(text).map(IdentityId::from)
285}
286
287fn parse_blob(text: &str) -> Option<Blob> {
288	Blob::from_hex(Fragment::internal(text)).ok()
289}
290
291fn parse_digest_text(inner: &ValueType, accuracy: u32, text: &str) -> Option<Digest> {
292	let bytes = decode(text.strip_prefix("0x")?).ok()?;
293	let digest = Digest::decode(&bytes).ok()?;
294	(*digest.inner() == *inner && digest.accuracy() == accuracy).then_some(digest)
295}
296
297fn parse_decimal_text(text: &str, precision: Precision, scale: Scale) -> Option<Decimal> {
298	let value = parse_decimal(Fragment::internal(text)).ok()?;
299	let unscaled = decimal_unscaled(&value, precision, scale)?;
300	Decimal::from_parts(unscaled, scale.value())
301}
302
303fn cells<T: Clone>(
304	name: &str,
305	ty: &ValueType,
306	rows: Vec<Option<String>>,
307	filler: T,
308	parse: impl Fn(&str) -> Option<T>,
309) -> Result<Vec<T>, DecodeError> {
310	rows.into_iter()
311		.enumerate()
312		.map(|(row, cell)| match cell {
313			None => Ok(filler.clone()),
314			Some(text) => parse(&text).ok_or_else(|| cell_error(name, row, ty, &text)),
315		})
316		.collect()
317}
318
319pub fn convert_column_to_data(
320	name: &str,
321	target: ValueType,
322	data: Vec<JsonValue>,
323) -> Result<(FieldRef, ArrayRef), DecodeError> {
324	let (base, depth) = peel_options(&target);
325	if depth > 1 {
326		return Err(DecodeError::ColumnDecodeFailed {
327			column_name: name.to_string(),
328			row_index: None,
329			source: Box::new(DecodeError::InvalidData(format!(
330				"type {target} has {depth} option layers, but a column holds at most one"
331			))),
332		});
333	}
334	if matches!(base, ValueType::List(_) | ValueType::Record(_)) {
335		return convert_list_or_record_column(name, &target, base, depth, data);
336	}
337	let data: Vec<String> = data
338		.into_iter()
339		.enumerate()
340		.map(|(row, payload)| match payload {
341			JsonValue::String(text) => Ok(text),
342			other => Err(column_error(
343				name,
344				row,
345				format!("expected a JSON string for {target}, got {other}"),
346			)),
347		})
348		.collect::<Result<_, _>>()?;
349	let mut layers = vec![vec![true; data.len()]; depth as usize];
350	let mut rows = Vec::with_capacity(data.len());
351	for (row, payload) in data.into_iter().enumerate() {
352		match none_marker_depth(&payload) {
353			None => rows.push(Some(payload)),
354			Some(wrapped) if wrapped < depth => {
355				for layer in &mut layers[wrapped as usize..] {
356					layer[row] = false;
357				}
358				rows.push(None);
359			}
360			Some(0) if matches!(base, ValueType::Digest { .. }) => rows.push(None),
361			Some(_) if depth == 0 => {
362				return Err(column_error(
363					name,
364					row,
365					format!("none marker for non-Option type {target}"),
366				));
367			}
368			Some(wrapped) => {
369				return Err(column_error(
370					name,
371					row,
372					format!(
373						"none marker depth {wrapped} exceeds the {depth} Option layers of type {target}"
374					),
375				));
376			}
377		}
378	}
379	let (value_type, array) = base_column(name, base, rows)?;
380	column(name, value_type, array, layers)
381}
382
383fn column(
384	name: &str,
385	value_type: ValueType,
386	array: ArrayRef,
387	layers: Vec<Vec<bool>>,
388) -> Result<(FieldRef, ArrayRef), DecodeError> {
389	decoded_column(name, value_type, array, layers.into_iter().next().map(BooleanBuffer::from), None).map_err(
390		|error| DecodeError::ColumnDecodeFailed {
391			column_name: name.to_string(),
392			row_index: None,
393			source: Box::new(error),
394		},
395	)
396}
397
398fn convert_list_or_record_column(
399	name: &str,
400	target: &ValueType,
401	base: &ValueType,
402	depth: u32,
403	data: Vec<JsonValue>,
404) -> Result<(FieldRef, ArrayRef), DecodeError> {
405	let mut layers = vec![vec![true; data.len()]; depth as usize];
406	let mut values = Vec::with_capacity(data.len());
407	for (row, payload) in data.into_iter().enumerate() {
408		match payload.as_str().and_then(none_marker_depth) {
409			None => {
410				let value = parse_json_value(base, &payload)
411					.map_err(|e| column_error(name, row, e.to_string()))?;
412				if let Value::None {
413					..
414				} = value
415				{
416					return Err(column_error(
417						name,
418						row,
419						format!("none for non-Option type {target}"),
420					));
421				}
422				values.push(Some(value));
423			}
424			Some(wrapped) if wrapped < depth => {
425				for layer in &mut layers[wrapped as usize..] {
426					layer[row] = false;
427				}
428				values.push(None);
429			}
430			Some(_) if depth == 0 => {
431				return Err(column_error(
432					name,
433					row,
434					format!("none marker for non-Option type {target}"),
435				));
436			}
437			Some(wrapped) => {
438				return Err(column_error(
439					name,
440					row,
441					format!(
442						"none marker depth {wrapped} exceeds the {depth} Option layers of type {target}"
443					),
444				));
445			}
446		}
447	}
448	column(name, base.clone(), Arc::new(any_array_optional(values)), layers)
449}
450
451fn base_column(name: &str, base: &ValueType, rows: Vec<Option<String>>) -> Result<(ValueType, ArrayRef), DecodeError> {
452	let array: ArrayRef = match base {
453		ValueType::Option(inner) => return base_column(name, inner, rows),
454		ValueType::Boolean => Arc::new(BooleanArray::from(cells(name, base, rows, false, |s| s.parse().ok())?)),
455		ValueType::Float4 => Arc::new(Float32Array::from(cells(name, base, rows, 0.0f32, |s| {
456			s.parse().ok().map(OrderedF32::canonical)
457		})?)),
458		ValueType::Float8 => Arc::new(Float64Array::from(cells(name, base, rows, 0.0f64, |s| {
459			s.parse().ok().map(OrderedF64::canonical)
460		})?)),
461		ValueType::Int1 => Arc::new(Int8Array::from(cells(name, base, rows, 0i8, |s| s.parse().ok())?)),
462		ValueType::Int2 => Arc::new(Int16Array::from(cells(name, base, rows, 0i16, |s| s.parse().ok())?)),
463		ValueType::Int4 => Arc::new(Int32Array::from(cells(name, base, rows, 0i32, |s| s.parse().ok())?)),
464		ValueType::Int8 => Arc::new(Int64Array::from(cells(name, base, rows, 0i64, |s| s.parse().ok())?)),
465		ValueType::Int16 => Arc::new(wide_array(cells(name, base, rows, 0i128, |s| s.parse().ok())?)),
466		ValueType::Uint1 => Arc::new(UInt8Array::from(cells(name, base, rows, 0u8, |s| s.parse().ok())?)),
467		ValueType::Uint2 => Arc::new(UInt16Array::from(cells(name, base, rows, 0u16, |s| s.parse().ok())?)),
468		ValueType::Uint4 => Arc::new(UInt32Array::from(cells(name, base, rows, 0u32, |s| s.parse().ok())?)),
469		ValueType::Uint8 => Arc::new(UInt64Array::from(cells(name, base, rows, 0u64, |s| s.parse().ok())?)),
470		ValueType::Uint16 => Arc::new(wide_array(cells(name, base, rows, 0u128, |s| s.parse().ok())?)),
471		ValueType::Date => Arc::new(date_array(cells(
472			name,
473			base,
474			rows,
475			Date::from_ymd(1970, 1, 1).unwrap(),
476			parse_date_text,
477		)?)),
478		ValueType::DateTime => Arc::new(datetime_array(cells(
479			name,
480			base,
481			rows,
482			DateTime::from_epoch_secs(0).unwrap(),
483			parse_datetime_text,
484		)?)),
485		ValueType::Time => Arc::new(time_array(cells(
486			name,
487			base,
488			rows,
489			Time::from_hms(0, 0, 0).unwrap(),
490			parse_time_text,
491		)?)),
492		ValueType::Duration => {
493			Arc::new(duration_array(cells(name, base, rows, Duration::zero(), parse_duration_text)?))
494		}
495		ValueType::Uuid4 => Arc::new(uuid4_array(cells(
496			name,
497			base,
498			rows,
499			parse_uuid4_text("00000000-0000-4000-8000-000000000000").unwrap(),
500			parse_uuid4_text,
501		)?)),
502		ValueType::Uuid7 => Arc::new(uuid7_array(cells(
503			name,
504			base,
505			rows,
506			parse_uuid7_text("00000000-0000-7000-8000-000000000000").unwrap(),
507			parse_uuid7_text,
508		)?)),
509		ValueType::IdentityId => Arc::new(identity_id_array(cells(
510			name,
511			base,
512			rows,
513			parse_identity_id("00000000-0000-7000-8000-000000000000").unwrap(),
514			parse_identity_id,
515		)?)),
516		ValueType::Blob => Arc::new(blob_array(&cells(name, base, rows, Blob::new(vec![]), parse_blob)?)),
517		ValueType::Decimal {
518			precision,
519			scale,
520		} => decimal_array(
521			*precision,
522			*scale,
523			cells(name, base, rows, Decimal::zero(), |text| parse_decimal_text(text, *precision, *scale))?,
524		)
525		.into_array(),
526		ValueType::Digest {
527			inner,
528			accuracy,
529		} => {
530			let mut builder = LargeBinaryBuilder::with_capacity(rows.len(), 0);
531			for (row, cell) in rows.into_iter().enumerate() {
532				match cell {
533					None => builder.append_null(),
534					Some(text) => {
535						let digest = parse_digest_text(inner, *accuracy, &text)
536							.ok_or_else(|| cell_error(name, row, base, &text))?;
537						push_digest(&mut builder, &digest);
538					}
539				}
540			}
541			Arc::new(builder.finish())
542		}
543		ValueType::Utf8
544		| ValueType::Any
545		| ValueType::DictionaryId
546		| ValueType::List(_)
547		| ValueType::Record(_)
548		| ValueType::Tuple(_) => {
549			let strings = cells(name, base, rows, String::new(), |s| Some(s.to_string()))?;
550			return Ok((ValueType::Utf8, Arc::new(LargeStringArray::from(strings))));
551		}
552		ValueType::Enum(_) => {
553			let strings = cells(name, base, rows, String::new(), |s| Some(s.to_string()))?;
554			Arc::new(LargeStringArray::from(strings))
555		}
556	};
557	Ok((base.clone(), array))
558}
reifydb_codec/json/from.rs

reifydb_codec/json/
from.rs
