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

reifydb_codec/json/
from.rs
