surrealdb_expr/val/

mod.rs

#![allow(clippy::derive_ord_xor_partial_ord)]

use std::cmp::Ordering;
use std::collections::{BTreeMap, HashMap};
use std::ops::Bound;
use std::time::Duration as StdDuration;

use anyhow::{Result, bail};
use chrono::{DateTime, Utc};
use common::fmt::{EscapeIdent, QuoteStr};
use geo::Point;
use revision::revisioned;
use rust_decimal::prelude::*;
use storekey::{BorrowDecode, Encode};
use surrealdb_collections::VecMap;
pub use surrealdb_strand::Strand;
use surrealdb_types::{SqlFormat, ToSql, write_sql};

use crate::expr;
use crate::expr::Error;
use crate::expr::kind::GeometryKind;
use crate::expr::statements::info::InfoStructure;
use crate::val::table_name_public::IntoPublicTable;

pub mod array;
pub mod bytes;
pub mod closure;
pub mod convert_public;
pub mod datetime;
pub mod duration;
pub mod evaluator;
pub mod field_path;
pub mod file;
pub mod geometry;
pub mod number;
pub mod object;
pub mod range;
pub mod record_id;
pub mod regex;
pub mod rnd;
pub mod set;
pub mod table_name_public;
pub mod uuid;
pub mod value;

pub use surrealdb_strand::TableName;

pub use self::array::Array;
pub use self::bytes::Bytes;
pub use self::closure::Closure;
pub use self::convert_public::convert_public_value_to_internal;
pub use self::datetime::Datetime;
pub use self::duration::Duration;
pub use self::evaluator::{BoxInvokeFut, ClosureEvaluator};
pub use self::file::File;
pub use self::geometry::Geometry;
pub use self::number::Number;
pub use self::object::Object;
pub use self::range::Range;
pub use self::record_id::{RecordId, RecordIdKey, RecordIdKeyRange, RecordIdentity};
pub use self::regex::Regex;
pub use self::set::Set;
pub use self::uuid::Uuid;
pub use self::value::{CastError, CoerceError};

/// Marker type for a different serialization format for value which does not encode type
/// information which is not required for indexing.
pub enum IndexFormat {}

/// Marker type for value conversions from Value::None
#[derive(Clone, Copy, Eq, PartialEq, PartialOrd)]
pub struct SqlNone;

/// Marker type for value conversions from Value::Null
#[derive(Clone, Copy, Eq, PartialEq, PartialOrd)]
pub struct Null;

// `revision(1)` is the legacy on-disk format (varint discriminant +
// payload). `revision(2, optimised)` wraps each value in a
// 1-byte tag (variant id + size class) so walkers can skip any value in
// O(1) via the `u32_le` length prefix on varlen variants.
//
// Optimised enum walkers are "materialised" — they don't expose
// `into_<variant>()` for streaming descent. Instead the macro emits
// `<variant>_view()` accessors that return `OwnedVariantView<T>` carrying
// a `Cow<'r, [u8]>` over the variant body. With the `WalkRevisioned:
// BorrowedReader` bound the source is slice-backed, so the Cow is
// borrowed — no `Vec<u8>` allocation per descent. The caller then
// constructs a fresh walker on the borrowed bytes:
//
//     let view = walker.object_view()?;          // zero-alloc borrow
//     let mut r: &[u8] = view.as_bytes();
//     let inner = Object::walk_revisioned(&mut r)?; // streaming Object walker
//
// This recovers fully streaming descent at the Value level while picking
// up the rev-2 wire compactness. Combined with rev-2 `Object`/`Array`/
// `Set` indexed prologues, the planner hot path gets O(log n) Map
// binary search, O(1) Array random access, and O(1) skip past any Value
// — all zero-alloc through the descent chain.
#[revisioned(revision(1), revision(2, optimised))]
#[derive(Clone, Debug, Default, PartialEq, PartialOrd, Hash, Encode, BorrowDecode)]
#[storekey(format = "()")]
#[storekey(format = "IndexFormat")]
pub enum Value {
	#[default]
	#[revision(size = "inline")]
	None,
	#[revision(size = "inline")]
	Null,
	#[revision(size = "fixed(1)")]
	Bool(bool),
	#[revision(size = "varlen")]
	Number(Number),
	#[revision(size = "varlen")]
	String(Strand),
	#[revision(size = "varlen")]
	Duration(Duration),
	#[revision(size = "varlen")]
	Datetime(Datetime),
	#[revision(size = "varlen")]
	Uuid(Uuid),
	#[revision(size = "varlen")]
	Array(Array),
	#[revision(size = "varlen")]
	Set(Set),
	#[revision(size = "varlen")]
	Object(Object),
	#[revision(size = "varlen")]
	Geometry(Geometry),
	#[revision(size = "varlen")]
	Bytes(Bytes),
	#[revision(size = "varlen")]
	Table(TableName),
	#[revision(size = "varlen")]
	RecordId(RecordId),
	#[revision(size = "varlen")]
	File(File),
	#[revision(size = "varlen")]
	Regex(Regex),
	#[revision(size = "varlen")]
	Range(Box<Range>),
	// Closure's manual `SerializeRevisioned` impl errors regardless of the
	// envelope. Mark `varlen` for completeness; the macro requires every
	// variant to declare a size class under `optimised`.
	#[revision(size = "varlen")]
	Closure(Box<Closure>),
	// Add new variants here
}

impl Eq for Value {}

impl Ord for Value {
	fn cmp(&self, other: &Self) -> Ordering {
		self.partial_cmp(other).unwrap_or(Ordering::Equal)
	}
}

impl Value {
	/// Whether a `HashMap`/`HashSet` **miss** on this value proves the collection
	/// does not hold it — i.e. whether this value's [`Hash`](std::hash::Hash) is
	/// consistent with its [`PartialEq`].
	///
	/// [`Number`] is the one leaf where the two disagree, and it cannot be
	/// repaired: see [`Number`]'s `Hash` impl for why. A value that can reach a
	/// [`Number`] may hash into a different bucket than an equal value, so
	/// `contains` answers `false` for something the collection holds.
	///
	/// For `==` a probe is decisive as soon as *either* side is number-free,
	/// because `==` requires matching variants at every level: two values can
	/// only hash apart while comparing equal if both carry a [`Number`] in the
	/// same position. So a caller judging `==` may trust a miss when this returns
	/// `true` for the probed value, *or* when it returns `true` for every element
	/// of the collection.
	///
	/// [`Value::equal`] — what `=`, `IN` and the `CONTAINS` family compare by —
	/// is wider than `==` in one place: it matches a top-level [`Value::String`]
	/// against a top-level [`Value::Regex`] by running the regex, and the two
	/// hash apart. [`Value::Regex`] reports `false` for that reason, although its
	/// own hash and `==` agree. A caller judging `equal` may trust a miss when
	/// this returns `true` on *both* sides. The either-side rule does not carry
	/// over, since a string reports `true`: it would trust a regex's miss against
	/// a set of the strings it matches, unless the caller rules that pairing out
	/// itself.
	///
	/// Variants whose hash/equality pairing is not established — [`Value::Geometry`]
	/// (hashes `f64` coordinates), [`Value::Range`] and [`Value::Closure`] —
	/// report `false` so they take the same conservative path.
	pub fn hash_agrees_with_eq(&self) -> bool {
		match self {
			Value::None
			| Value::Null
			| Value::Bool(_)
			| Value::String(_)
			| Value::Duration(_)
			| Value::Datetime(_)
			| Value::Uuid(_)
			| Value::Bytes(_)
			| Value::Table(_)
			| Value::File(_) => true,
			Value::Number(_)
			| Value::Geometry(_)
			| Value::Regex(_)
			| Value::Range(_)
			| Value::Closure(_) => false,
			Value::Array(a) => a.iter().all(Value::hash_agrees_with_eq),
			Value::Set(s) => s.iter().all(Value::hash_agrees_with_eq),
			// An `Object`'s equality covers its keys too, but those are `Strand`
			// (hash is `as_str().hash`), so only the values can carry a `Number`.
			Value::Object(o) => o.values().all(Value::hash_agrees_with_eq),
			Value::RecordId(r) => r.key.hash_agrees_with_eq(),
		}
	}

	pub const NONE: Self = Self::None;

	// -----------------------------------
	// Initial record value
	// -----------------------------------

	/// Create an empty Object Value
	pub fn empty_object() -> Self {
		Value::Object(Object::default())
	}

	// -----------------------------------
	// Simple value detection
	// -----------------------------------

	/// Check if this Value is not NONE
	pub fn is_some(&self) -> bool {
		!matches!(self, Value::None)
	}

	/// Check if this Value is NONE or NULL
	pub fn is_nullish(&self) -> bool {
		matches!(self, Value::None | Value::Null)
	}

	/// Check if this Value is NONE
	pub fn is_empty_array(&self) -> bool {
		if let Value::Array(v) = self {
			v.is_empty()
		} else {
			false
		}
	}

	/// Check if this Value is truthy
	pub fn is_truthy(&self) -> bool {
		match self {
			Value::Bool(v) => *v,
			Value::Uuid(_) => true,
			Value::RecordId(_) => true,
			Value::Geometry(_) => true,
			Value::Datetime(_) => true,
			Value::Bytes(v) => !v.is_empty(),
			Value::Array(v) => !v.is_empty(),
			Value::Set(v) => !v.is_empty(),
			Value::Object(v) => !v.is_empty(),
			Value::String(v) => !v.is_empty(),
			Value::Number(v) => v.is_truthy(),
			Value::Duration(v) => v.as_nanos() > 0,
			_ => false,
		}
	}

	/// Check if this Value is an int Number
	pub fn is_int(&self) -> bool {
		matches!(self, Value::Number(Number::Int(_)))
	}

	/// Check if this Value is a float Number
	pub fn is_float(&self) -> bool {
		matches!(self, Value::Number(Number::Float(_)))
	}

	/// Check if this Value is a decimal Number
	pub fn is_decimal(&self) -> bool {
		matches!(self, Value::Number(Number::Decimal(_)))
	}

	/// Check if this Value is a RecordId of a specific type
	pub fn is_record_type(&self, types: &[TableName]) -> bool {
		match self {
			Value::RecordId(v) => v.is_table_type(types),
			_ => false,
		}
	}

	/// Check if this Value is a Geometry of a specific type
	pub fn is_geometry_type(&self, types: &[GeometryKind]) -> bool {
		match self {
			Value::Geometry(Geometry::Point(_)) => {
				types.iter().any(|t| matches!(t, GeometryKind::Point))
			}
			Value::Geometry(Geometry::Line(_)) => {
				types.iter().any(|t| matches!(t, GeometryKind::Line))
			}
			Value::Geometry(Geometry::Polygon(_)) => {
				types.iter().any(|t| matches!(t, GeometryKind::Polygon))
			}
			Value::Geometry(Geometry::MultiPoint(_)) => {
				types.iter().any(|t| matches!(t, GeometryKind::MultiPoint))
			}
			Value::Geometry(Geometry::MultiLine(_)) => {
				types.iter().any(|t| matches!(t, GeometryKind::MultiLine))
			}
			Value::Geometry(Geometry::MultiPolygon(_)) => {
				types.iter().any(|t| matches!(t, GeometryKind::MultiPolygon))
			}
			Value::Geometry(Geometry::Collection(_)) => {
				types.iter().any(|t| matches!(t, GeometryKind::Collection))
			}
			_ => false,
		}
	}

	// -----------------------------------
	// Simple conversion of value
	// -----------------------------------

	/// Converts this Value into an unquoted String
	pub fn into_raw_string(self) -> String {
		match self {
			Value::String(v) => v.into_string(),
			Value::Uuid(v) => v.to_string(),
			Value::Datetime(v) => v.to_string(),
			_ => self.to_sql(),
		}
	}

	/// Converts this Value into an unquoted String
	pub fn to_raw_string(&self) -> String {
		match self {
			Value::String(v) => v.as_str().to_owned(),
			Value::Uuid(v) => v.to_string(),
			Value::Datetime(v) => v.to_string(),
			_ => self.to_sql(),
		}
	}

	/// Returns the surql representation of the kind of the value as a string.
	///
	/// # Warning
	/// This function is not fully implement for all variants, make sure you
	/// don't accidentally use it where it can return an invalid value.
	pub fn kind_of(&self) -> &'static str {
		match self {
			Self::None => "none",
			Self::Null => "null",
			Self::Bool(_) => "bool",
			Self::Uuid(_) => "uuid",
			Self::Array(_) => "array",
			Self::Set(_) => "set",
			Self::Object(_) => "object",
			Self::String(_) => "string",
			Self::Duration(_) => "duration",
			Self::Datetime(_) => "datetime",
			Self::Closure(_) => "function",
			Self::Number(Number::Int(_)) => "int",
			Self::Number(Number::Float(_)) => "float",
			Self::Number(Number::Decimal(_)) => "decimal",
			Self::Geometry(Geometry::Point(_)) => "geometry<point>",
			Self::Geometry(Geometry::Line(_)) => "geometry<line>",
			Self::Geometry(Geometry::Polygon(_)) => "geometry<polygon>",
			Self::Geometry(Geometry::MultiPoint(_)) => "geometry<multipoint>",
			Self::Geometry(Geometry::MultiLine(_)) => "geometry<multiline>",
			Self::Geometry(Geometry::MultiPolygon(_)) => "geometry<multipolygon>",
			Self::Geometry(Geometry::Collection(_)) => "geometry<collection>",
			Self::Regex(_) => "regex",
			Self::File(_) => "file",
			Self::Bytes(_) => "bytes",
			Self::Range(_) => "range",
			Self::RecordId(_) => "record",
			Self::Table(_) => "table",
		}
	}

	// -----------------------------------
	// Value operations
	// -----------------------------------

	/// Check if this Value is equal to another Value
	pub fn equal(&self, other: &Value) -> bool {
		match self {
			Value::None => other.is_none(),
			Value::Null => other.is_null(),
			Value::Bool(v) => match other {
				Value::Bool(w) => v == w,
				_ => false,
			},
			Value::Uuid(v) => match other {
				Value::Uuid(w) => v == w,
				_ => false,
			},
			Value::RecordId(v) => match other {
				Value::RecordId(w) => v == w,
				_ => false,
			},
			Value::String(v) => match other {
				Value::String(w) => v == w,
				Value::Regex(w) => w.inner().is_match(v.as_str()),
				_ => false,
			},
			Value::Regex(v) => match other {
				Value::Regex(w) => v == w,
				Value::String(w) => v.inner().is_match(w.as_str()),
				_ => false,
			},
			Value::Array(v) => match other {
				Value::Array(w) => v == w,
				_ => false,
			},
			Value::Set(v) => match other {
				Value::Set(w) => v == w,
				_ => false,
			},
			Value::Object(v) => match other {
				Value::Object(w) => v == w,
				_ => false,
			},
			Value::Number(v) => match other {
				Value::Number(w) => v == w,
				_ => false,
			},
			Value::Geometry(v) => match other {
				Value::Geometry(w) => v == w,
				_ => false,
			},
			Value::Duration(v) => match other {
				Value::Duration(w) => v == w,
				_ => false,
			},
			Value::Datetime(v) => match other {
				Value::Datetime(w) => v == w,
				_ => false,
			},
			_ => self == other,
		}
	}

	/// Check if all Values in an Array are equal to another Value
	pub fn all_equal(&self, other: &Value) -> bool {
		match self {
			Value::Array(v) => v.iter().all(|v| v.equal(other)),
			Value::Set(v) => v.iter().all(|v| v.equal(other)),
			_ => self.equal(other),
		}
	}

	/// Check if any Values in an Array are equal to another Value
	pub fn any_equal(&self, other: &Value) -> bool {
		match self {
			Value::Array(v) => v.iter().any(|v| v.equal(other)),
			Value::Set(v) => v.iter().any(|v| v.equal(other)),
			_ => self.equal(other),
		}
	}

	/// Check if this Value contains another Value
	pub fn contains(&self, other: &Value) -> bool {
		match self {
			Value::Array(v) => v.iter().any(|v| v.equal(other)),
			Value::Set(v) => v.contains(other),
			Value::Uuid(v) => match other {
				Value::String(w) => v.to_string().contains(w.as_str()),
				_ => false,
			},
			Value::String(v) => match other {
				Value::String(w) => v.contains(w.as_str()),
				_ => false,
			},
			Value::Geometry(v) => match other {
				Value::Geometry(w) => v.contains(w),
				_ => false,
			},
			Value::Object(v) => match other {
				Value::String(w) => v.contains_key(w.as_str()),
				_ => false,
			},
			Value::Range(r) => {
				let beg = match &r.start {
					Bound::Unbounded => true,
					Bound::Included(beg) => beg.le(other),
					Bound::Excluded(beg) => beg.lt(other),
				};

				beg && match &r.end {
					Bound::Unbounded => true,
					Bound::Included(end) => end.ge(other),
					Bound::Excluded(end) => end.gt(other),
				}
			}
			_ => false,
		}
	}

	/// Check if all Values in an Array, Set or String contain another Value
	pub fn contains_all(&self, other: &Value) -> bool {
		match other {
			Value::Array(v) if v.iter().all(|v| v.is_strand()) && self.is_strand() => {
				// confirmed as strand so all return false is unreachable
				let Value::String(this) = self else {
					return false;
				};
				v.iter().all(|s| {
					let Value::String(other_string) = s else {
						return false;
					};
					this.contains(&**other_string)
				})
			}
			Value::Set(v) if v.iter().all(|v| v.is_strand()) && self.is_strand() => {
				// confirmed as strand so all return false is unreachable
				let Value::String(this) = self else {
					return false;
				};
				v.iter().all(|s| {
					let Value::String(other_string) = s else {
						return false;
					};
					this.contains(&**other_string)
				})
			}
			Value::Array(v) => v.iter().all(|v| match self {
				Value::Array(w) => w.iter().any(|w| v.equal(w)),
				Value::Set(w) => w.iter().any(|w| v.equal(w)),
				Value::Geometry(_) => self.contains(v),
				_ => false,
			}),
			Value::Set(v) => v.iter().all(|v| match self {
				Value::Array(w) => w.iter().any(|w| v.equal(w)),
				Value::Set(w) => w.iter().any(|w| v.equal(w)),
				Value::Geometry(_) => self.contains(v),
				_ => false,
			}),
			Value::String(other_strand) => match self {
				Value::String(s) => s.contains(&**other_strand),
				_ => false,
			},
			_ => false,
		}
	}

	/// Check if any Values in an Array, Set or String contain another Value
	pub fn contains_any(&self, other: &Value) -> bool {
		match other {
			Value::Array(v) if v.iter().all(|v| v.is_strand()) && self.is_strand() => {
				// confirmed as strand so all return false is unreachable
				let Value::String(this) = self else {
					return false;
				};
				v.iter().any(|s| {
					let Value::String(other_string) = s else {
						return false;
					};
					this.contains(&**other_string)
				})
			}
			Value::Set(v) if v.iter().all(|v| v.is_strand()) && self.is_strand() => {
				// confirmed as strand so all return false is unreachable
				let Value::String(this) = self else {
					return false;
				};
				v.iter().any(|s| {
					let Value::String(other_string) = s else {
						return false;
					};
					this.contains(&**other_string)
				})
			}
			Value::Array(v) => v.iter().any(|v| match self {
				Value::Array(w) => w.iter().any(|w| v.equal(w)),
				Value::Set(w) => w.iter().any(|w| v.equal(w)),
				Value::Geometry(_) => self.contains(v),
				_ => false,
			}),
			Value::Set(v) => v.iter().any(|v| match self {
				Value::Array(w) => w.iter().any(|w| v.equal(w)),
				Value::Set(w) => w.iter().any(|w| v.equal(w)),
				Value::Geometry(_) => self.contains(v),
				_ => false,
			}),
			Value::String(other_strand) => match self {
				Value::String(s) => s.contains(&**other_strand),
				_ => false,
			},
			_ => false,
		}
	}

	/// Check if this Geometry intersects another Value
	pub fn intersects(&self, other: &Value) -> bool {
		match self {
			Value::Geometry(v) => match other {
				Value::Geometry(w) => v.intersects(w),
				_ => false,
			},
			_ => false,
		}
	}

	// -----------------------------------
	// Sorting operations
	// -----------------------------------

	/// Compare this Value to another Value lexicographically
	pub fn lexical_cmp(&self, other: &Value) -> Option<Ordering> {
		match (self, other) {
			(Value::String(a), Value::String(b)) => Some(lexicmp::lexical_cmp(a, b)),
			_ => self.partial_cmp(other),
		}
	}

	/// Compare this Value to another Value using natural numerical comparison
	pub fn natural_cmp(&self, other: &Value) -> Option<Ordering> {
		match (self, other) {
			(Value::String(a), Value::String(b)) => Some(lexicmp::natural_cmp(a, b)),
			_ => self.partial_cmp(other),
		}
	}

	/// Compare this Value to another Value lexicographically and using natural
	/// numerical comparison
	pub fn natural_lexical_cmp(&self, other: &Value) -> Option<Ordering> {
		match (self, other) {
			(Value::String(a), Value::String(b)) => Some(lexicmp::natural_lexical_cmp(a, b)),
			_ => self.partial_cmp(other),
		}
	}

	/// Turns this value into a literal evaluating to the same value.
	pub fn into_literal(self) -> expr::Expr {
		match self {
			Value::None => expr::Expr::Literal(expr::Literal::None),
			Value::Null => expr::Expr::Literal(expr::Literal::Null),
			Value::Bool(x) => expr::Expr::Literal(expr::Literal::Bool(x)),
			Value::Number(Number::Int(i)) => expr::Expr::Literal(expr::Literal::Integer(i)),
			Value::Number(Number::Float(f)) => expr::Expr::Literal(expr::Literal::Float(f)),
			Value::Number(Number::Decimal(d)) => expr::Expr::Literal(expr::Literal::Decimal(d)),
			Value::String(strand) => expr::Expr::Literal(expr::Literal::String(strand)),
			Value::Duration(duration) => expr::Expr::Literal(expr::Literal::Duration(duration)),
			Value::Datetime(datetime) => expr::Expr::Literal(expr::Literal::Datetime(datetime)),
			Value::Uuid(uuid) => expr::Expr::Literal(expr::Literal::Uuid(uuid)),
			Value::Array(array) => expr::Expr::Literal(expr::Literal::Array(array.into_literal())),
			Value::Set(set) => {
				// Convert set to array for literal representation since there's no set literal
				// syntax
				expr::Expr::Literal(expr::Literal::Array(set.into_literal()))
			}
			Value::Object(object) => {
				expr::Expr::Literal(expr::Literal::Object(object.into_literal()))
			}
			Value::Geometry(geometry) => expr::Expr::Literal(expr::Literal::Geometry(geometry)),
			Value::Bytes(bytes) => expr::Expr::Literal(expr::Literal::Bytes(bytes)),
			Value::RecordId(record_id) => {
				expr::Expr::Literal(expr::Literal::RecordId(record_id.into_literal()))
			}
			Value::Regex(regex) => expr::Expr::Literal(expr::Literal::Regex(regex)),
			Value::File(file) => expr::Expr::Literal(expr::Literal::File(file)),
			Value::Closure(closure) => closure.into_expr(),
			Value::Range(range) => range.into_literal(),
			Value::Table(t) => expr::Expr::Table(t),
		}
	}
}

impl ToSql for Value {
	fn fmt_sql(&self, f: &mut String, sql_fmt: SqlFormat) {
		match self {
			Value::None => f.push_str("NONE"),
			Value::Null => f.push_str("NULL"),
			Value::Bool(v) => v.fmt_sql(f, sql_fmt),
			Value::Number(v) => v.fmt_sql(f, sql_fmt),
			Value::String(v) => write_sql!(f, sql_fmt, "{}", QuoteStr(v.as_str())),
			Value::Duration(v) => v.fmt_sql(f, sql_fmt),
			Value::Datetime(v) => v.fmt_sql(f, sql_fmt),
			Value::Uuid(v) => v.fmt_sql(f, sql_fmt),
			Value::Array(v) => v.fmt_sql(f, sql_fmt),
			Value::Set(v) => v.fmt_sql(f, sql_fmt),
			Value::Object(v) => v.fmt_sql(f, sql_fmt),
			Value::Geometry(v) => v.fmt_sql(f, sql_fmt),
			Value::Bytes(v) => v.fmt_sql(f, sql_fmt),
			// Escaped explicitly rather than through the name's own `ToSql`: the
			// engine's `TableName` deliberately has none, and it derefs to `str`,
			// so a bare `fmt_sql` here would silently resolve to the unescaped
			// string impl and emit a name that cannot be read back.
			Value::Table(v) => EscapeIdent(v.as_str()).fmt_sql(f, sql_fmt),
			Value::RecordId(v) => v.fmt_sql(f, sql_fmt),
			Value::File(v) => v.fmt_sql(f, sql_fmt),
			Value::Regex(v) => v.fmt_sql(f, sql_fmt),
			Value::Range(v) => v.fmt_sql(f, sql_fmt),
			Value::Closure(v) => v.fmt_sql(f, sql_fmt),
		}
	}
}

impl InfoStructure for Value {
	fn structure(self) -> Value {
		self.to_sql().into()
	}
}

// ------------------------------

pub trait TryAdd<Rhs = Self> {
	type Output;
	fn try_add(self, rhs: Rhs) -> Result<Self::Output>;
}

use std::ops::Add;

impl TryAdd for Value {
	type Output = Self;
	fn try_add(self, other: Self) -> Result<Self> {
		Ok(match (self, other) {
			(Self::Number(v), Self::Number(w)) => Self::Number(v.try_add(w)?),
			(Self::String(v), Self::String(w)) => {
				let mut s = String::with_capacity(v.len() + w.len());
				s.push_str(v.as_str());
				s.push_str(w.as_str());
				Value::String(s.into())
			}
			(Self::Datetime(v), Self::Duration(w)) => Self::Datetime(w.try_add(v)?),
			(Self::Duration(v), Self::Datetime(w)) => Self::Datetime(v.try_add(w)?),
			(Self::Duration(v), Self::Duration(w)) => Self::Duration(v.try_add(w)?),
			(Self::Array(v), Self::Array(w)) => Self::Array(v.concat(w)),
			(Self::Array(v), Self::Set(w)) => Self::Array(v.concat_set(w)),

			(Self::Set(mut v), Self::Set(w)) => {
				v.0.extend(w.0);
				Self::Set(v)
			}
			(Self::Set(mut v), Self::Array(w)) => {
				v.0.extend(w.0);
				Self::Set(v)
			}
			(Self::Object(v), Self::Object(w)) => Self::Object(v.add(w)),
			(v, w) => bail!(Error::TryAdd(v.kind_of().to_owned(), w.kind_of().to_owned())),
		})
	}
}

// ------------------------------

pub trait TrySub<Rhs = Self> {
	type Output;
	fn try_sub(self, v: Rhs) -> Result<Self::Output>;
}

impl TrySub for Value {
	type Output = Self;
	fn try_sub(self, other: Self) -> Result<Self> {
		Ok(match (self, other) {
			(Self::Number(v), Self::Number(w)) => Self::Number(v.try_sub(w)?),
			(Self::Datetime(v), Self::Datetime(w)) => Self::Duration(v.try_sub(w)?),
			(Self::Datetime(v), Self::Duration(w)) => Self::Datetime(w.try_sub(v)?),
			(Self::Duration(v), Self::Datetime(w)) => Self::Datetime(v.try_sub(w)?),
			(Self::Duration(v), Self::Duration(w)) => Self::Duration(v.try_sub(w)?),
			(Self::Array(v), Self::Array(x)) => Self::from(v.remove_all(&x.0)),
			(Self::Array(v), Self::Set(x)) => Self::from(v.remove_all_set(&x)),
			(Self::Set(mut v), Self::Array(x)) => {
				for item in x.0 {
					v.remove(&item);
				}
				Self::from(v)
			}
			(Self::Set(mut v), Self::Set(x)) => {
				for item in x {
					v.remove(&item);
				}
				Self::from(v)
			}
			(v, w) => bail!(Error::TrySub(v.kind_of().to_owned(), w.kind_of().to_owned())),
		})
	}
}

// ------------------------------

pub trait TryMul<Rhs = Self> {
	type Output;
	fn try_mul(self, v: Self) -> Result<Self::Output>;
}

impl TryMul for Value {
	type Output = Self;
	fn try_mul(self, other: Self) -> Result<Self> {
		match (self, other) {
			(Self::Number(v), Self::Number(w)) => Ok(Self::Number(v.try_mul(w)?)),
			(Self::Duration(d), Self::Number(Number::Int(i))) => {
				if i < 0 {
					bail!("Cannot multiply a duration with a negative number");
				}

				let factor: u32 = i.try_into()?;

				let res =
					d.0.checked_mul(factor)
						.ok_or_else(|| Error::ArithmeticOverflow(format!("{d} * {i}")))?;

				Ok(Value::Duration(Duration(res)))
			}
			(Self::Duration(d), Self::Number(Number::Float(f))) => {
				if !(f.is_finite() && f >= 0.0) {
					bail!("Cannot multiply a duration with a non-finite or negative number");
				}

				let secs = d.0.as_secs_f64() * f;
				let res = StdDuration::try_from_secs_f64(secs)
					.map_err(|_| Error::ArithmeticOverflow(format!("{d} * {f}")))?;

				Ok(Value::Duration(Duration(res)))
			}
			(Self::Duration(d), Self::Number(Number::Decimal(dec))) => {
				if !dec.is_sign_positive() && !dec.is_zero() {
					bail!("Cannot multiply a duration with a negative number");
				}

				let nanos: Decimal = d.nanos().into();
				let scaled = (nanos * dec).trunc();

				let scaled_nanos: u128 = scaled
					.to_u128()
					.ok_or_else(|| Error::ArithmeticOverflow(format!("{d} * {dec}")))?;

				let secs: u64 = (scaled_nanos / 1_000_000_000)
					.try_into()
					.map_err(|_| Error::ArithmeticOverflow(format!("{d} * {dec}")))?;

				let subsec_nanos: u32 = (scaled_nanos % 1_000_000_000).try_into()?;

				Ok(Value::Duration(Duration(StdDuration::new(secs, subsec_nanos))))
			}
			(v, w) => bail!(Error::TryMul(v.kind_of().to_owned(), w.kind_of().to_owned())),
		}
	}
}

// ------------------------------

pub trait TryDiv<Rhs = Self> {
	type Output;
	fn try_div(self, v: Self) -> Result<Self::Output>;
}

impl TryDiv for Value {
	type Output = Self;
	fn try_div(self, other: Self) -> Result<Self> {
		match (self, other) {
			(Self::Number(v), Self::Number(w)) => Ok(Self::Number(v.try_div(w)?)),
			(Self::Duration(d), Self::Number(Number::Int(i))) => {
				if i <= 0 {
					bail!("A duration can only be divided by a value greater than 0.");
				}

				let denom: u32 = i.try_into()?;
				let res = d.0 / denom;
				Ok(Value::Duration(Duration(res)))
			}
			(Self::Duration(d), Self::Number(Number::Float(f))) => {
				if !(f.is_finite() && f > 0.0) {
					bail!("A duration can only be divided by a finite value greater than 0.");
				}
				let secs = d.as_secs_f64() / f;

				let res = StdDuration::try_from_secs_f64(secs)?;
				Ok(Value::Duration(Duration(res)))
			}
			(Self::Duration(d), Self::Number(Number::Decimal(dec))) => {
				if !dec.is_sign_positive() || dec.is_zero() {
					bail!("A duration can only be divided by a value greater than 0.");
				}

				let nanos: Decimal = d.nanos().into();

				let divided = (nanos / dec).trunc();

				let divided_nanos: u128 = divided
					.to_u128()
					.ok_or_else(|| anyhow::anyhow!("Resulting duration is out of range"))?;

				let secs: u64 = (divided_nanos / 1_000_000_000)
					.try_into()
					.map_err(|_| anyhow::anyhow!("Resulting duration seconds out of range"))?;

				let subsec_nanos: u32 = (divided_nanos % 1_000_000_000).try_into()?;

				Ok(Value::Duration(Duration(StdDuration::new(secs, subsec_nanos))))
			}

			(v, w) => bail!(Error::TryDiv(v.kind_of().to_owned(), w.kind_of().to_owned())),
		}
	}
}

// ------------------------------

pub trait TryFloatDiv<Rhs = Self> {
	type Output;
	fn try_float_div(self, v: Self) -> Result<Self::Output>;
}

impl TryFloatDiv for Value {
	type Output = Self;
	fn try_float_div(self, other: Self) -> Result<Self::Output> {
		Ok(match (self, other) {
			(Self::Number(v), Self::Number(w)) => Self::Number(v.try_float_div(w)?),
			(v, w) => bail!(Error::TryDiv(v.kind_of().to_owned(), w.kind_of().to_owned())),
		})
	}
}

// ------------------------------

pub trait TryRem<Rhs = Self> {
	type Output;
	fn try_rem(self, v: Self) -> Result<Self::Output>;
}

impl TryRem for Value {
	type Output = Self;
	fn try_rem(self, other: Self) -> Result<Self> {
		Ok(match (self, other) {
			(Self::Number(v), Self::Number(w)) => Self::Number(v.try_rem(w)?),
			(v, w) => bail!(Error::TryRem(v.kind_of().to_owned(), w.kind_of().to_owned())),
		})
	}
}

// ------------------------------

pub trait TryPow<Rhs = Self> {
	type Output;
	fn try_pow(self, v: Self) -> Result<Self::Output>;
}

impl TryPow for Value {
	type Output = Self;
	fn try_pow(self, other: Self) -> Result<Self> {
		Ok(match (self, other) {
			(Value::Number(v), Value::Number(w)) => Self::Number(v.try_pow(w)?),
			(v, w) => bail!(Error::TryPow(v.kind_of().to_owned(), w.kind_of().to_owned())),
		})
	}
}

// ------------------------------

pub trait TryNeg<Rhs = Self> {
	type Output;
	fn try_neg(self) -> Result<Self::Output>;
}

impl TryNeg for Value {
	type Output = Self;
	fn try_neg(self) -> Result<Self> {
		Ok(match self {
			Self::Number(n) => Self::Number(n.try_neg()?),
			v => bail!(Error::TryNeg(v.kind_of().to_owned())),
		})
	}
}

// Conversion methods.

/// Macro implementing conversion methods for the variants of the value enum.
macro_rules! subtypes {
	($($name:ident$( ( $($t:tt)* ) )? => ($is:ident,$as:ident,$into:ident)),*$(,)?) => {
		impl Value {
			$(
				subtypes!{@method $name $( ($($t)*) )? => $is,$as,$into}
			)*
		}

			$(
				subtypes!{@from $name $( ($($t)*) )? => $is,$as,$into}
			)*

	};

	(@pat $name:ident($t:ty)) => {
		Value::$name(_)
	};

	(@pat $name:ident) => {
		Value::$name
	};

	(@method $name:ident($t:ty) => $is:ident,$as:ident,$into:ident) => {
		#[doc = concat!("Check if the value is a [`",stringify!($name),"`]")]
		#[allow(dead_code)]
		pub fn $is(&self) -> bool{
			matches!(self,Value::$name(_))
		}

		#[doc = concat!("Return a reference to [`",stringify!($name),"`] if the value is of that type")]
		#[allow(dead_code)]
		pub fn $as(&self) -> Option<&$t>{
			if let Value::$name(x) = self{
				Some(x)
			}else{
				None
			}
		}

		#[doc = concat!("Turns the value into a [`",stringify!($name),"`] returning None if the value is not of that type")]
		#[allow(dead_code)]
		pub fn $into(self) -> Option<$t>{
			if let Value::$name(x) = self{
				Some(x)
			}else{
				None
			}
		}
	};

	(@method $name:ident => $is:ident,$as:ident,$into:ident) => {
		#[doc = concat!("Check if the value is a [`",stringify!($name),"`]")]
		#[allow(dead_code)]
		pub fn $is(&self) -> bool{
			matches!(self,Value::$name)
		}
	};


	(@from $name:ident(Box<$inner:ident>) => $is:ident,$as:ident,$into:ident) => {
		impl From<$inner> for Value {
			fn from(v: $inner) -> Self{
				Value::$name(Box::new(v))
			}
		}

		impl From<Box<$inner>> for Value {
			fn from(v: Box<$inner>) -> Self{
				Value::$name(v)
			}
		}
	};

	(@from $name:ident($t:ident) => $is:ident,$as:ident,$into:ident) => {
		impl From<$t> for Value {
			fn from(v: $t) -> Self{
				Value::$name(v)
			}
		}
	};

	(@from $name:ident => $is:ident,$as:ident,$into:ident) => {
		// skip
	};

}

subtypes! {
	None => (is_none,_unused,_unused),
	Null => (is_null,_unused,_unused),
	Bool(bool) => (is_bool,as_bool,into_bool),
	Number(Number) => (is_number,as_number,into_number),
	String(Strand) => (is_strand,as_strand,into_strand),
	Table(TableName) => (is_table,as_table,into_table),
	Duration(Duration) => (is_duration,as_duration,into_duration),
	Datetime(Datetime) => (is_datetime,as_datetime,into_datetime),
	Uuid(Uuid) => (is_uuid,as_uuid,into_uuid),
	Array(Array) => (is_array,as_array,into_array),
	Set(Set) => (is_set,as_set,into_set),
	Object(Object) => (is_object,as_object,into_object),
	Geometry(Geometry) => (is_geometry,as_geometry,into_geometry),
	Bytes(Bytes) => (is_bytes,as_bytes,into_bytes),
	RecordId(RecordId) => (is_record,as_record,into_record),
	Regex(Regex) => (is_regex,as_regex,into_regex),
	Range(Box<Range>) => (is_range,as_range,into_range),
	Closure(Box<Closure>) => (is_closure,as_closure,into_closure),
	File(File) => (is_file,as_file,into_file),
}

macro_rules! impl_from_number {
	($($n:ident),*$(,)?) => {
		$(
			impl From<$n> for Value{
				fn from(v: $n) -> Self{
					Value::Number(Number::from(v))
				}
			}
		)*
	};
}
impl_from_number!(i8, i16, i32, i64, u8, u16, u32, isize, f32, f64, Decimal);

impl From<Vec<Value>> for Value {
	fn from(value: Vec<Value>) -> Self {
		Value::Array(Array(value))
	}
}

impl From<Null> for Value {
	fn from(_v: Null) -> Self {
		Value::Null
	}
}

// `u64`/`usize` always fit losslessly into a Number (as an Int, or a Decimal
// when they exceed i64), so these conversions are infallible.
impl From<u64> for Value {
	fn from(v: u64) -> Self {
		Value::Number(Number::from(v))
	}
}

impl From<usize> for Value {
	fn from(v: usize) -> Self {
		Value::Number(Number::from(v))
	}
}

// `i128`/`u128` can exceed Decimal's range, so these conversions are fallible
// and error rather than truncate when the value is too large to represent.
impl TryFrom<i128> for Value {
	type Error = Error;
	fn try_from(v: i128) -> Result<Self, Self::Error> {
		Ok(Value::Number(Number::try_from(v)?))
	}
}

impl TryFrom<u128> for Value {
	type Error = Error;
	fn try_from(v: u128) -> Result<Self, Self::Error> {
		Ok(Value::Number(Number::try_from(v)?))
	}
}

impl From<&str> for Value {
	fn from(v: &str) -> Self {
		Self::String(Strand::from(v))
	}
}

impl From<String> for Value {
	fn from(v: String) -> Self {
		Self::String(Strand::from(v))
	}
}

impl From<::uuid::Uuid> for Value {
	fn from(v: ::uuid::Uuid) -> Self {
		Value::Uuid(Uuid::from(v))
	}
}

impl From<DateTime<Utc>> for Value {
	fn from(v: DateTime<Utc>) -> Self {
		Value::Datetime(Datetime::from(v))
	}
}

impl From<Point<f64>> for Value {
	fn from(v: Point<f64>) -> Self {
		Value::Geometry(Geometry::from(v))
	}
}

impl From<HashMap<&str, Value>> for Value {
	fn from(v: HashMap<&str, Value>) -> Self {
		Value::Object(Object::from(v))
	}
}

impl From<HashMap<String, Value>> for Value {
	fn from(v: HashMap<String, Value>) -> Self {
		Value::Object(Object::from(v))
	}
}

impl From<BTreeMap<String, Value>> for Value {
	fn from(v: BTreeMap<String, Value>) -> Self {
		Value::Object(Object::from(v))
	}
}

impl From<VecMap<String, Value>> for Value {
	fn from(v: VecMap<String, Value>) -> Self {
		Value::Object(Object::from(v))
	}
}

impl From<VecMap<Strand, Value>> for Value {
	fn from(v: VecMap<Strand, Value>) -> Self {
		Value::Object(Object(v))
	}
}

impl From<VecMap<&str, Value>> for Value {
	fn from(v: VecMap<&str, Value>) -> Self {
		Value::Object(Object::from(v))
	}
}

impl From<BTreeMap<&str, Value>> for Value {
	fn from(v: BTreeMap<&str, Value>) -> Self {
		Value::Object(Object::from(v))
	}
}

impl TryFrom<Value> for crate::types::PublicValue {
	type Error = anyhow::Error;

	fn try_from(s: Value) -> Result<Self, Self::Error> {
		convert_value_to_public_value(s)
	}
}

impl From<crate::types::PublicValue> for Value {
	fn from(s: crate::types::PublicValue) -> Self {
		convert_public_value_to_internal(s)
	}
}

impl FromIterator<Value> for Value {
	fn from_iter<I: IntoIterator<Item = Value>>(iter: I) -> Self {
		Value::Array(Array(iter.into_iter().collect()))
	}
}

impl FromIterator<(String, Value)> for Value {
	fn from_iter<I: IntoIterator<Item = (String, Value)>>(iter: I) -> Self {
		Value::Object(Object::from_iter(iter))
	}
}

/// Convert our internal value `crate::val::Value` to the public value `surrealdb_types::Value`.
///
/// In the future, as the two types diverge, this function will need access to the context in order
/// to convert certain values to the public value.
/// Maximum nesting depth a [`Value`] may carry.
///
/// Several code paths walk a value with native recursion — the public-value
/// conversion every query result passes through, the SQL formatter used by
/// export and `INFO`, and the derived `revision` storage codec. None can carry
/// its own cumulative depth counter across the others, and `surrealdb-core` is
/// built with `panic = "abort"` under the release profile, so a stack overflow
/// in any of them is an unconditional process abort rather than a failed
/// request.
///
/// The parser's own caps do not bound this: they are per-parse, and a value is
/// grown one level per request by a read-modify-write loop
/// (`UPDATE deep:1 SET v = [v]`) that stays far under them at every step. The
/// compute cap (`max_computation_depth`) bounds expression nesting, not the
/// runtime nesting of a value being cloned or serialised.
///
/// Sized well above any legitimate document and far below the depth at which
/// the recursive walks approach a thread stack.
pub const MAX_VALUE_DEPTH: usize = 256;

/// A node in the iterative depth walk. Nesting reaches a `Value` through several
/// wrappers that are not themselves `Value`s - a record id's key, a
/// `GeometryCollection`, and a closure's captured variables - so the worklist
/// carries any of them.
enum DepthNode<'a> {
	Value(&'a Value),
	Key(&'a crate::val::RecordIdKey),
	Geo(&'a crate::val::Geometry),
	Closure(&'a crate::val::Closure),
}

/// Rejects a [`Value`] nested deeper than [`MAX_VALUE_DEPTH`].
///
/// Deliberately iterative: a recursive checker would overflow the stack on
/// exactly the input it exists to reject. Traversal stops at the first node
/// past the cap, so a deeply-nested value is rejected without being walked in
/// full.
///
/// A record id's key is walked too, and a geometry's `GeometryCollection`
/// nesting. The derived storage codec, the SQL formatter and the recursive drop
/// all descend into a compound key (`t:[[[...]]]`) and into a nested collection,
/// so anything deep enough to overflow them must be rejected at the same point a
/// deep field value is.
pub fn check_value_depth(value: &Value) -> Result<()> {
	use crate::val::RecordIdKey;
	// Fast path: a value that nests nothing is depth 1 and cannot exceed the
	// cap, so skip allocating the traversal stack. This keeps the check
	// allocation-free for the common scalar case - e.g. a `fold` / `reduce`
	// accumulator that reduces to a number, which is checked every iteration.
	// `Geometry` (a `GeometryCollection` nests) and `Closure` (its captures hold
	// values) are containers here.
	if !matches!(
		value,
		Value::Array(_)
			| Value::Set(_)
			| Value::Object(_)
			| Value::Range(_)
			| Value::RecordId(_)
			| Value::Geometry(_)
			| Value::Closure(_)
	) {
		return Ok(());
	}
	// (node, depth of that node)
	let mut stack: Vec<(DepthNode<'_>, usize)> = vec![(DepthNode::Value(value), 1)];
	while let Some((node, depth)) = stack.pop() {
		if depth > MAX_VALUE_DEPTH {
			bail!(Error::ValueTooDeep {
				max: MAX_VALUE_DEPTH,
			});
		}
		match node {
			DepthNode::Value(value) => match value {
				Value::Array(v) => {
					stack.extend(v.0.iter().map(|v| (DepthNode::Value(v), depth + 1)))
				}
				Value::Set(v) => stack.extend(v.0.iter().map(|v| (DepthNode::Value(v), depth + 1))),
				Value::Object(v) => {
					stack.extend(v.0.values().map(|v| (DepthNode::Value(v), depth + 1)))
				}
				Value::Range(v) => {
					for bound in [&v.start, &v.end] {
						match bound {
							Bound::Included(v) | Bound::Excluded(v) => {
								stack.push((DepthNode::Value(v), depth + 1))
							}
							Bound::Unbounded => {}
						}
					}
				}
				Value::RecordId(v) => stack.push((DepthNode::Key(&v.key), depth + 1)),
				// A geometry and a closure nest within themselves, not within
				// `Value`, so the wrapper is pushed at `depth` (it is not itself a
				// level) and recurses via its own node below.
				Value::Geometry(g) => stack.push((DepthNode::Geo(g), depth)),
				Value::Closure(c) => stack.push((DepthNode::Closure(c), depth)),
				// Leaves, listed explicitly so a new nesting `Value` variant is a
				// compile error here rather than a silent hole.
				Value::None
				| Value::Null
				| Value::Bool(_)
				| Value::Number(_)
				| Value::String(_)
				| Value::Duration(_)
				| Value::Datetime(_)
				| Value::Uuid(_)
				| Value::Bytes(_)
				| Value::Table(_)
				| Value::File(_)
				| Value::Regex(_) => {}
			},
			DepthNode::Key(key) => match key {
				RecordIdKey::Array(a) => {
					stack.extend(a.0.iter().map(|v| (DepthNode::Value(v), depth + 1)))
				}
				RecordIdKey::Object(o) => {
					stack.extend(o.0.values().map(|v| (DepthNode::Value(v), depth + 1)))
				}
				RecordIdKey::Range(r) => {
					for bound in [&r.start, &r.end] {
						match bound {
							Bound::Included(k) | Bound::Excluded(k) => {
								stack.push((DepthNode::Key(k), depth + 1))
							}
							Bound::Unbounded => {}
						}
					}
				}
				RecordIdKey::Number(_) | RecordIdKey::String(_) | RecordIdKey::Uuid(_) => {}
			},
			DepthNode::Geo(geo) => {
				// Only a `GeometryCollection` nests further geometries; the
				// point/line/polygon variants are bounded-depth primitives.
				if let crate::val::Geometry::Collection(geoms) = geo {
					stack.extend(geoms.iter().map(|g| (DepthNode::Geo(g), depth + 1)));
				}
			}
			DepthNode::Closure(closure) => {
				// A closure captures the values it references, so a closure
				// capturing a closure nests. The `body` / `args` are parse-bounded
				// AST, not runtime values; only the captures grow. A `Builtin`
				// closure (the compiled form the streaming engine produces) holds
				// its captures behind the trait, so walk them through
				// `captured_values` rather than treating it as a leaf.
				match closure {
					crate::val::Closure::Expr {
						captures,
						..
					} => {
						stack.extend(captures.0.values().map(|v| (DepthNode::Value(v), depth + 1)));
					}
					crate::val::Closure::Builtin(builtin) => {
						stack.extend(
							builtin
								.captured_values()
								.into_iter()
								.map(|v| (DepthNode::Value(v), depth + 1)),
						);
					}
				}
			}
		}
	}
	Ok(())
}

pub fn convert_value_to_public_value(value: crate::val::Value) -> Result<surrealdb_types::Value> {
	convert_value_to_public_value_at(value, 1)
}

/// Converts a row-shaped statement result (the array of rows a non-`ONLY`
/// `SELECT` and friends emit) to a public value, treating the enclosing array as
/// a transport envelope rather than a nesting level: each row is converted as
/// its own root at depth 1.
///
/// This is the anchor the storage write guard uses - `check_value_depth` runs on
/// a record's value with the record at depth 1 - and the one the streaming read
/// path already uses (it converts each row as it is emitted). Converting the
/// whole result array at depth 1 instead would charge every row an extra level,
/// so a record whose deepest field sits exactly at [`MAX_VALUE_DEPTH`] would
/// store and stream fine yet fail a collected read. A `Scalar` result (a single
/// value, e.g. `SELECT ... ONLY` or `RETURN`) is its own root and must go
/// through [`convert_value_to_public_value`] instead.
pub fn convert_result_rows_to_public(value: crate::val::Value) -> Result<surrealdb_types::Value> {
	let crate::val::Value::Array(rows) = value else {
		// A row-shaped result is always the array of emitted rows; any other
		// shape is converted as its own root (defensive - not expected).
		return convert_value_to_public_value(value);
	};
	let converted =
		rows.0.into_iter().map(convert_value_to_public_value).collect::<Result<Vec<_>>>()?;
	Ok(surrealdb_types::Value::Array(surrealdb_types::Array::from(converted)))
}

/// [`convert_value_to_public_value`] carrying the depth of `value`.
///
/// The depth is threaded through the mutual recursion rather than checked by a
/// separate pass so the universal query-result path pays only a counter, and so
/// a value already in storage from before [`MAX_VALUE_DEPTH`] was enforced
/// surfaces an error here instead of aborting the process.
fn convert_value_to_public_value_at(
	value: crate::val::Value,
	depth: usize,
) -> Result<surrealdb_types::Value> {
	if depth > MAX_VALUE_DEPTH {
		bail!(Error::ValueTooDeep {
			max: MAX_VALUE_DEPTH,
		});
	}
	match value {
		crate::val::Value::None => Ok(surrealdb_types::Value::None),
		crate::val::Value::Null => Ok(surrealdb_types::Value::Null),
		crate::val::Value::Bool(value) => Ok(surrealdb_types::Value::Bool(value)),
		crate::val::Value::Number(value) => convert_number_to_public(value),
		crate::val::Value::String(value) => Ok(surrealdb_types::Value::String(value.into())),
		crate::val::Value::Datetime(value) => convert_datetime_to_public(value),
		crate::val::Value::Duration(value) => convert_duration_to_public(value),
		crate::val::Value::Uuid(value) => convert_uuid_to_public(value),
		crate::val::Value::Array(value) => convert_array_to_public(value, depth),
		crate::val::Value::Set(value) => convert_set_to_public(value, depth),
		crate::val::Value::Object(value) => convert_object_to_public(value, depth),
		crate::val::Value::Geometry(value) => convert_geometry_to_public(value, depth),
		crate::val::Value::Bytes(value) => convert_bytes_to_public(value),
		crate::val::Value::RecordId(value) => convert_record_id_to_public(value, depth),
		crate::val::Value::File(value) => convert_file_to_public(value),
		crate::val::Value::Range(value) => convert_range_to_public(*value, depth),
		crate::val::Value::Regex(value) => convert_regex_to_public(value),
		crate::val::Value::Table(value) => {
			Ok(surrealdb_types::Value::Table(value.into_public_table()))
		}
		crate::val::Value::Closure(_) => {
			Err(anyhow::anyhow!("Closure values cannot be converted to public value"))
		}
	}
}

fn convert_number_to_public(value: crate::val::Number) -> Result<surrealdb_types::Value> {
	let number = match value {
		crate::val::Number::Int(i) => surrealdb_types::Number::Int(i),
		crate::val::Number::Float(f) => surrealdb_types::Number::Float(f),
		crate::val::Number::Decimal(d) => surrealdb_types::Number::Decimal(d),
	};
	Ok(surrealdb_types::Value::Number(number))
}

fn convert_datetime_to_public(value: crate::val::Datetime) -> Result<surrealdb_types::Value> {
	Ok(surrealdb_types::Value::Datetime(surrealdb_types::Datetime::from(value.0)))
}

fn convert_duration_to_public(value: crate::val::Duration) -> Result<surrealdb_types::Value> {
	Ok(surrealdb_types::Value::Duration(surrealdb_types::Duration::from_std(value.0)))
}

fn convert_uuid_to_public(value: crate::val::Uuid) -> Result<surrealdb_types::Value> {
	Ok(surrealdb_types::Value::Uuid(surrealdb_types::Uuid::from(value.0)))
}

fn convert_bytes_to_public(value: crate::val::Bytes) -> Result<surrealdb_types::Value> {
	Ok(surrealdb_types::Value::Bytes(surrealdb_types::Bytes::from(value.0)))
}

fn convert_regex_to_public(value: crate::val::Regex) -> Result<surrealdb_types::Value> {
	Ok(surrealdb_types::Value::Regex(surrealdb_types::Regex::from(value.0)))
}

fn convert_file_to_public(value: crate::val::File) -> Result<surrealdb_types::Value> {
	Ok(surrealdb_types::Value::File(surrealdb_types::File::new(value.bucket, value.key)))
}

fn convert_geometry_to_public(
	value: crate::val::Geometry,
	depth: usize,
) -> Result<surrealdb_types::Value> {
	// A `GeometryCollection` nests, so carry the depth through it as the other
	// container arms do; the storage guard (`check_value_depth`) rejects a deep
	// geometry at write, and this keeps the read path in step.
	if depth > MAX_VALUE_DEPTH {
		bail!(Error::ValueTooDeep {
			max: MAX_VALUE_DEPTH,
		});
	}
	use surrealdb_types::Geometry as PublicGeometry;
	let geometry = match value {
		crate::val::Geometry::Point(p) => PublicGeometry::Point(p),
		crate::val::Geometry::Line(l) => PublicGeometry::Line(l),
		crate::val::Geometry::Polygon(p) => PublicGeometry::Polygon(p),
		crate::val::Geometry::MultiPoint(mp) => PublicGeometry::MultiPoint(mp),
		crate::val::Geometry::MultiLine(ml) => PublicGeometry::MultiLine(ml),
		crate::val::Geometry::MultiPolygon(mp) => PublicGeometry::MultiPolygon(mp),
		crate::val::Geometry::Collection(c) => {
			let converted: Result<Vec<_>> = c
				.into_iter()
				.map(|g| {
					if let surrealdb_types::Value::Geometry(g) =
						convert_geometry_to_public(g, depth + 1)?
					{
						Ok(g)
					} else {
						Err(anyhow::anyhow!("Failed to convert geometry collection item"))
					}
				})
				.collect();
			PublicGeometry::Collection(converted?)
		}
	};
	Ok(surrealdb_types::Value::Geometry(geometry))
}

fn convert_array_to_public(
	value: crate::val::Array,
	depth: usize,
) -> Result<surrealdb_types::Value> {
	let converted: Result<Vec<_>> =
		value.0.into_iter().map(|v| convert_value_to_public_value_at(v, depth + 1)).collect();
	Ok(surrealdb_types::Value::Array(surrealdb_types::Array::from(converted?)))
}

fn convert_set_to_public(value: crate::val::Set, depth: usize) -> Result<surrealdb_types::Value> {
	// Set is stored as a sorted Vec internally
	let converted: Result<Vec<surrealdb_types::Value>> =
		value.0.into_iter().map(|v| convert_value_to_public_value_at(v, depth + 1)).collect();
	Ok(surrealdb_types::Value::Set(surrealdb_types::Set::from(converted?)))
}

fn convert_object_to_public(
	value: crate::val::Object,
	depth: usize,
) -> Result<surrealdb_types::Value> {
	let converted = convert_object_to_public_map_at(value, depth)?;
	Ok(surrealdb_types::Value::Object(surrealdb_types::Object::from(converted)))
}

pub fn convert_object_to_public_map(
	value: crate::val::Object,
) -> Result<BTreeMap<String, surrealdb_types::Value>> {
	convert_object_to_public_map_at(value, 1)
}

fn convert_object_to_public_map_at(
	value: crate::val::Object,
	depth: usize,
) -> Result<BTreeMap<String, surrealdb_types::Value>> {
	value
		.0
		.into_iter()
		.map(|(k, v)| convert_value_to_public_value_at(v, depth + 1).map(|v| (k.into_string(), v)))
		.collect()
}

fn convert_record_id_to_public(
	value: crate::val::RecordId,
	depth: usize,
) -> Result<surrealdb_types::Value> {
	// A record id's key nests one level under it, matching `check_value_depth`'s
	// `Key` node at `depth + 1`, so continue the count there. Without the `+ 1`
	// the read path accepts ~2x the record-id nesting the storage guard rejects.
	let key = convert_record_id_key_to_public(value.key, depth + 1)?;
	Ok(surrealdb_types::Value::RecordId(surrealdb_types::RecordId {
		table: value.table.into_public_table(),
		key,
	}))
}

fn convert_record_id_key_to_public(
	key: crate::val::RecordIdKey,
	depth: usize,
) -> Result<surrealdb_types::RecordIdKey> {
	if depth > MAX_VALUE_DEPTH {
		bail!(Error::ValueTooDeep {
			max: MAX_VALUE_DEPTH,
		});
	}
	match key {
		crate::val::RecordIdKey::Number(n) => Ok(surrealdb_types::RecordIdKey::Number(n)),
		crate::val::RecordIdKey::String(s) => Ok(surrealdb_types::RecordIdKey::String(s.into())),
		crate::val::RecordIdKey::Uuid(u) => {
			Ok(surrealdb_types::RecordIdKey::Uuid(surrealdb_types::Uuid::from(u.0)))
		}
		crate::val::RecordIdKey::Array(a) => {
			let converted_array = convert_array_to_public(a, depth)?;
			if let surrealdb_types::Value::Array(arr) = converted_array {
				Ok(surrealdb_types::RecordIdKey::Array(arr))
			} else {
				Err(anyhow::anyhow!("Failed to convert record id key array"))
			}
		}
		crate::val::RecordIdKey::Object(o) => {
			let converted_object = convert_object_to_public(o, depth)?;
			if let surrealdb_types::Value::Object(obj) = converted_object {
				Ok(surrealdb_types::RecordIdKey::Object(obj))
			} else {
				Err(anyhow::anyhow!("Failed to convert record id key object"))
			}
		}
		crate::val::RecordIdKey::Range(r) => {
			Ok(surrealdb_types::RecordIdKey::Range(Box::new(surrealdb_types::RecordIdKeyRange {
				start: match r.start {
					Bound::Included(k) => {
						Bound::Included(convert_record_id_key_to_public(k, depth + 1)?)
					}
					Bound::Excluded(k) => {
						Bound::Excluded(convert_record_id_key_to_public(k, depth + 1)?)
					}
					Bound::Unbounded => Bound::Unbounded,
				},
				end: match r.end {
					Bound::Included(k) => {
						Bound::Included(convert_record_id_key_to_public(k, depth + 1)?)
					}
					Bound::Excluded(k) => {
						Bound::Excluded(convert_record_id_key_to_public(k, depth + 1)?)
					}
					Bound::Unbounded => Bound::Unbounded,
				},
			})))
		}
	}
}

fn convert_range_to_public(
	value: crate::val::Range,
	depth: usize,
) -> Result<surrealdb_types::Value> {
	// A range's bounds nest one level under it, so continue the depth count
	// through them (`depth + 1`) as the array/object arms do. This keeps the read
	// path in agreement with `check_value_depth`, which also descends into range
	// bounds, so a bound whose value sits at the cap makes the range one past it
	// and is rejected rather than converted through a deeper recursion.
	Ok(surrealdb_types::Value::Range(Box::new(surrealdb_types::Range {
		start: match value.start {
			Bound::Included(v) => Bound::Included(convert_value_to_public_value_at(v, depth + 1)?),
			Bound::Excluded(v) => Bound::Excluded(convert_value_to_public_value_at(v, depth + 1)?),
			Bound::Unbounded => Bound::Unbounded,
		},
		end: match value.end {
			Bound::Included(v) => Bound::Included(convert_value_to_public_value_at(v, depth + 1)?),
			Bound::Excluded(v) => Bound::Excluded(convert_value_to_public_value_at(v, depth + 1)?),
			Bound::Unbounded => Bound::Unbounded,
		},
	})))
}

#[cfg(test)]
mod tests {
	use super::{MAX_VALUE_DEPTH, check_value_depth, convert_value_to_public_value};
	use crate::val::{Array, Object, Value};

	/// Builds a value nested `depth` levels deep: `depth` of 1 is a bare
	/// integer, 2 is `[1]`, 3 is `[[1]]`, and so on.
	fn nested_array(depth: usize) -> Value {
		let mut value = Value::from(1i64);
		for _ in 1..depth {
			value = Value::Array(Array::from(vec![value]));
		}
		value
	}

	fn nested_object(depth: usize) -> Value {
		let mut value = Value::from(1i64);
		for _ in 1..depth {
			let mut obj = Object::default();
			obj.insert("v".to_owned(), value);
			value = Value::Object(obj);
		}
		value
	}

	// The recursion sinks that walk a `Value` — the public-value conversion, the
	// SQL formatter, and the derived storage codec — carry no depth guard of
	// their own, and a stack overflow in any of them aborts the process under
	// the release profile's `panic = "abort"`. A value is grown one level per
	// request by a read-modify-write loop, so the parser's per-parse caps never
	// fire during accumulation; the depth of the value itself has to be bounded.
	#[test]
	fn value_depth_within_the_cap_is_accepted() {
		check_value_depth(&nested_array(MAX_VALUE_DEPTH))
			.expect("a value exactly at the cap must be accepted");
		check_value_depth(&nested_object(MAX_VALUE_DEPTH))
			.expect("objects nest the same way as arrays");
		check_value_depth(&Value::from(1i64)).expect("a scalar is depth 1");
	}

	#[test]
	fn value_depth_past_the_cap_is_rejected() {
		let err = check_value_depth(&nested_array(MAX_VALUE_DEPTH + 1))
			.expect_err("a value one level past the cap must be rejected");
		assert!(err.to_string().contains("nested too deeply"), "unexpected error: {err}");
		assert!(check_value_depth(&nested_object(MAX_VALUE_DEPTH + 1)).is_err());
	}

	// The check must survive the input it exists to reject: a recursive
	// implementation would overflow the stack rather than return an error. The
	// depth used is far past any thread stack's tolerance for a recursive walk,
	// so this test aborts rather than fails if the check regresses to recursion.
	//
	// The value is leaked deliberately. `Value`'s derived `Drop` is itself
	// recursive, so releasing a value this deep overflows the stack whatever
	// this check does — which is the reason the depth cap is enforced when a
	// record is written rather than only where the recursive walks are: a value
	// past the cap must never come to exist, because merely dropping one is
	// fatal.
	#[test]
	fn value_depth_check_does_not_recurse() {
		let deep = std::mem::ManuallyDrop::new(nested_array(200_000));
		assert!(
			check_value_depth(&deep).is_err(),
			"a very deeply nested value must be rejected, not overflow the stack"
		);
	}

	// The conversion every query result passes through carries the depth
	// through its own mutual recursion, so a value already in storage from
	// before the cap was enforced surfaces an error rather than aborting.
	#[test]
	fn public_value_conversion_rejects_values_past_the_cap() {
		convert_value_to_public_value(nested_array(MAX_VALUE_DEPTH))
			.expect("a value at the cap must still convert");
		let err = convert_value_to_public_value(nested_array(MAX_VALUE_DEPTH + 1))
			.expect_err("a value past the cap must not convert");
		assert!(err.to_string().contains("nested too deeply"), "unexpected error: {err}");
	}

	// A collected row-shaped result is an array of rows, each its own root at the
	// depth the storage guard used (a record at depth 1). Converting the whole
	// result array charges every row an extra level, so a row at exactly the cap
	// fails that way but must convert when treated as a row - the collected-read
	// half of the storage/read agreement.
	#[test]
	fn result_rows_conversion_does_not_charge_the_envelope() {
		use super::convert_result_rows_to_public;

		let row = nested_array(MAX_VALUE_DEPTH);
		let result = Value::Array(Array::from(vec![row]));

		// Whole-array conversion puts the row at depth 2 and rejects it.
		assert!(
			convert_value_to_public_value(result.clone()).is_err(),
			"the envelope-charging path must reject a row at the cap"
		);
		// Per-row conversion accepts it, matching storage and the streaming path.
		convert_result_rows_to_public(result).expect("a row at the cap must convert as a row");
	}

	// A range's bounds nest one level under it, so the conversion carries the
	// depth through them: a bound holding a value already at the cap makes the
	// whole value one past it and must be rejected rather than converted through
	// a deeper recursion.
	#[test]
	fn public_value_conversion_threads_depth_through_ranges() {
		use std::ops::Bound;

		use crate::val::Range;

		let range_bound = |v: Value| {
			Value::Range(Box::new(Range {
				start: Bound::Included(v),
				end: Bound::Unbounded,
			}))
		};

		// A shallow value inside a range still converts.
		convert_value_to_public_value(range_bound(nested_array(8)))
			.expect("a shallow range-nested value must convert");
		// A range whose bound is at the cap makes the whole value one past it.
		let err = convert_value_to_public_value(range_bound(nested_array(MAX_VALUE_DEPTH)))
			.expect_err("a range-nested value past the cap must not convert");
		assert!(err.to_string().contains("nested too deeply"), "unexpected error: {err}");
	}

	/// A `GeometryCollection` nests further geometries, and both the SQL
	/// formatter and the derived `Drop` recurse into it, so the depth guard must
	/// walk it rather than treat a `Geometry` as a leaf, and the conversion must
	/// carry the depth through it.
	#[test]
	fn geometry_collection_depth_is_walked() {
		use crate::val::Geometry;

		// `depth` collections deep, innermost an empty collection.
		fn nested_geometry(depth: usize) -> Value {
			let mut g = Geometry::Collection(vec![]);
			for _ in 1..depth {
				g = Geometry::Collection(vec![g]);
			}
			Value::Geometry(g)
		}

		// The `Value::Geometry` wrapper and its outermost collection share a level
		// (the geometry nests within itself, not within `Value`), so `depth`
		// collections deep is exactly `depth` levels: the cap is the boundary for
		// both the walk and the conversion, and they must agree to the level.
		check_value_depth(&nested_geometry(MAX_VALUE_DEPTH))
			.expect("a geometry collection exactly at the cap must be accepted");
		convert_value_to_public_value(nested_geometry(MAX_VALUE_DEPTH))
			.expect("a geometry collection at the cap must also convert");
		let err = check_value_depth(&nested_geometry(MAX_VALUE_DEPTH + 1))
			.expect_err("a geometry one level past the cap must be rejected");
		assert!(err.to_string().contains("nested too deeply"), "unexpected error: {err}");
		let err = convert_value_to_public_value(nested_geometry(MAX_VALUE_DEPTH + 1))
			.expect_err("a geometry one level past the cap must not convert");
		assert!(err.to_string().contains("nested too deeply"), "unexpected error: {err}");

		// The walk stays iterative: a geometry too deep to drop recursively is
		// still rejected rather than overflowing the checker. Leaked because its
		// own `Drop` is recursive.
		let deep = std::mem::ManuallyDrop::new(nested_geometry(200_000));
		assert!(
			check_value_depth(&deep).is_err(),
			"a very deep geometry must be rejected, not overflow the stack"
		);
	}

	/// A record id nests through its key, which the storage codec and the
	/// recursive drop both descend into, so the key must be walked like any other
	/// nested value rather than treated as a leaf. The walk and the conversion
	/// must agree on the boundary: a record id counts one level for the id itself
	/// and one for the key array before its nested contents, so a key holding a
	/// value nested `MAX_VALUE_DEPTH - 2` levels sits exactly at the cap.
	#[test]
	fn record_id_key_depth_is_walked() {
		use crate::val::{RecordId, RecordIdKey};

		// A record id whose key wraps an array nested `inner` levels.
		fn record_id_with_nested_key(inner: usize) -> Value {
			Value::RecordId(RecordId {
				table: "t".into(),
				key: RecordIdKey::Array(Array::from(vec![nested_array(inner)])),
			})
		}

		check_value_depth(&record_id_with_nested_key(MAX_VALUE_DEPTH - 2))
			.expect("a compound key exactly at the cap must be accepted");
		convert_value_to_public_value(record_id_with_nested_key(MAX_VALUE_DEPTH - 2))
			.expect("a compound key at the cap must also convert");
		let err = check_value_depth(&record_id_with_nested_key(MAX_VALUE_DEPTH - 1))
			.expect_err("a record id whose key nests one past the cap must be rejected");
		assert!(err.to_string().contains("nested too deeply"), "unexpected error: {err}");
		let err = convert_value_to_public_value(record_id_with_nested_key(MAX_VALUE_DEPTH - 1))
			.expect_err("a record id whose key nests one past the cap must not convert");
		assert!(err.to_string().contains("nested too deeply"), "unexpected error: {err}");
	}

	/// `array::fold` can thread a closure-valued accumulator that captures itself
	/// each turn (`|$a, $_| |$x| $a`), growing the closure's capture nesting
	/// without bound, so the walk must descend into a closure's captures rather
	/// than treat it as a leaf. A closure cannot be converted to a public value
	/// or stored, so the walk the fold cap runs is the only guard on this shape.
	#[test]
	fn closure_capture_depth_is_walked() {
		use crate::expr::Expr;
		use crate::expr::variables::Variables;
		use crate::val::Closure;

		// A closure whose single capture holds a value nested `inner` levels. The
		// closure shares a level with its `Value` wrapper and the capture nests
		// one under it, so a capture at `MAX_VALUE_DEPTH - 1` sits at the cap.
		fn closure_with_nested_capture(inner: usize) -> Value {
			let mut captures = Variables::default();
			captures.insert("a", nested_array(inner));
			Value::Closure(Box::new(Closure::Expr {
				args: vec![],
				returns: None,
				body: Expr::Break,
				captures,
			}))
		}

		check_value_depth(&closure_with_nested_capture(MAX_VALUE_DEPTH - 1))
			.expect("a closure whose capture sits at the cap must be accepted");
		let err = check_value_depth(&closure_with_nested_capture(MAX_VALUE_DEPTH))
			.expect_err("a closure whose capture nests past the cap must be rejected");
		assert!(err.to_string().contains("nested too deeply"), "unexpected error: {err}");

		// The walk stays iterative on a closure too deep to drop recursively.
		let deep = std::mem::ManuallyDrop::new(closure_with_nested_capture(200_000));
		assert!(
			check_value_depth(&deep).is_err(),
			"a very deep closure capture must be rejected, not overflow the stack"
		);
	}

	/// The streaming engine represents a runtime closure as a `Closure::Builtin`
	/// whose captures live behind the `BuiltinClosureObj` trait, not a
	/// `Closure::Expr`, so the walk must reach them through `captured_values` or a
	/// closure-valued accumulator escapes the cap on that engine. The captures
	/// nest exactly as `Closure::Expr`'s do, so the boundary is the same.
	#[test]
	fn builtin_closure_captures_are_walked() {
		use std::sync::Arc;

		use crate::val::Closure;
		use crate::val::closure::BuiltinClosureObj;

		// A builtin closure that captured `values` by cloning them, the shape a
		// compiled closure takes.
		#[derive(Debug)]
		struct Captured(Vec<Value>);
		impl BuiltinClosureObj for Captured {
			fn as_any(&self) -> &dyn std::any::Any {
				self
			}
			fn captured_values(&self) -> Vec<&Value> {
				self.0.iter().collect()
			}
		}
		fn builtin_capturing(inner: usize) -> Value {
			Value::Closure(Box::new(Closure::Builtin(Arc::new(Captured(vec![nested_array(
				inner,
			)])))))
		}

		check_value_depth(&builtin_capturing(MAX_VALUE_DEPTH - 1))
			.expect("a builtin closure whose capture sits at the cap must be accepted");
		let err = check_value_depth(&builtin_capturing(MAX_VALUE_DEPTH))
			.expect_err("a builtin closure whose capture nests past the cap must be rejected");
		assert!(err.to_string().contains("nested too deeply"), "unexpected error: {err}");
	}

	/// A table name that collides with a keyword must render quoted, or the
	/// output cannot be parsed back. `TableName` derefs to `str`, so the
	/// rendering here cannot rely on method resolution finding an escaping
	/// impl — this pins the escaping itself rather than which impl provides it.
	/// `Number: Hash` disagrees with `Number: PartialEq`, so any value that can
	/// reach a number must report that a hash miss on it proves nothing.
	#[test]
	fn hash_agrees_with_eq_flags_numbers_at_any_depth() {
		use std::collections::BTreeMap;

		use crate::val::{Number, Object, RecordId, RecordIdKey, Value};

		let num = Value::Number(Number::Float(0.1));
		let nested =
			|v: Value| Value::Object(Object::from(BTreeMap::from([(Strand::from("v"), v)])));

		for v in [
			Value::None,
			Value::Null,
			Value::Bool(true),
			Value::String("x".into()),
			Value::from(vec![Value::String("x".into())]),
			nested(Value::String("x".into())),
			Value::RecordId(RecordId {
				table: "t".into(),
				key: RecordIdKey::Number(1),
			}),
		] {
			assert!(v.hash_agrees_with_eq(), "{v:?}");
		}

		for v in [
			num.clone(),
			Value::from(vec![num.clone()]),
			nested(num.clone()),
			nested(Value::from(vec![num.clone()])),
			Value::RecordId(RecordId {
				table: "t".into(),
				key: RecordIdKey::Array(vec![num].into()),
			}),
		] {
			assert!(!v.hash_agrees_with_eq(), "{v:?}");
		}
	}

	#[test]
	fn table_value_renders_escaped() {
		use surrealdb_types::ToSql;

		use crate::val::{TableName, Value};

		assert_eq!(Value::Table(TableName::from("users")).to_sql(), "users");
		assert_eq!(Value::Table(TableName::from("table")).to_sql(), "`table`");
		assert_eq!(Value::Table(TableName::from("a b")).to_sql(), "`a b`");
	}
	use chrono::{TimeZone, Utc};
	use geo::{MultiLineString, MultiPoint, MultiPolygon, line_string, point, polygon};
	use rstest::rstest;
	use rust_decimal::Decimal;
	use serde_json::{Value as Json, json};

	use super::*;
	use crate::syn;
	use crate::types::{
		PublicArray, PublicBytes, PublicDatetime, PublicDuration, PublicGeometry, PublicNumber,
		PublicObject, PublicRecordId, PublicRecordIdKey, PublicUuid, PublicValue,
	};
	use crate::val::Uuid;

	macro_rules! parse_val {
		($input:expr) => {
			crate::val::convert_public_value_to_internal(syn::value($input).unwrap())
		};
	}

	#[test]
	fn check_none() {
		assert!(Value::None.is_none());
		assert!(!Value::Null.is_none());
		assert!(!Value::from(1).is_none());
	}

	#[test]
	fn check_null() {
		assert!(Value::Null.is_null());
		assert!(!Value::None.is_null());
		assert!(!Value::from(1).is_null());
	}

	#[test]
	fn convert_truthy() {
		assert!(!Value::None.is_truthy());
		assert!(!Value::Null.is_truthy());
		assert!(Value::Bool(true).is_truthy());
		assert!(!Value::Bool(false).is_truthy());
		assert!(!Value::from(0).is_truthy());
		assert!(Value::from(1).is_truthy());
		assert!(Value::from(-1).is_truthy());
		assert!(Value::from(1.1).is_truthy());
		assert!(Value::from(-1.1).is_truthy());
		assert!(Value::from("true").is_truthy());
		assert!(Value::from("false").is_truthy());
		assert!(Value::from("falsey").is_truthy());
		assert!(Value::from("something").is_truthy());
		assert!(Value::from(Uuid::nil()).is_truthy());
		assert!(Value::from(Utc.with_ymd_and_hms(1948, 12, 3, 0, 0, 0).unwrap()).is_truthy());
	}

	#[test]
	fn convert_string() {
		assert_eq!(String::from("NONE"), Value::None.into_raw_string());
		assert_eq!(String::from("NULL"), Value::Null.into_raw_string());
		assert_eq!(String::from("true"), Value::Bool(true).into_raw_string());
		assert_eq!(String::from("false"), Value::Bool(false).into_raw_string());
		assert_eq!(String::from("0"), Value::from(0).into_raw_string());
		assert_eq!(String::from("1"), Value::from(1).into_raw_string());
		assert_eq!(String::from("-1"), Value::from(-1).into_raw_string());
		assert_eq!(String::from("1.1f"), Value::from(1.1).into_raw_string());
		assert_eq!(String::from("-1.1f"), Value::from(-1.1).into_raw_string());
		assert_eq!(String::from("3"), Value::from("3").into_raw_string());
		assert_eq!(String::from("true"), Value::from("true").into_raw_string());
		assert_eq!(String::from("false"), Value::from("false").into_raw_string());
		assert_eq!(String::from("something"), Value::from("something").into_raw_string());
	}

	#[test]
	fn check_size() {
		assert!(64 >= std::mem::size_of::<Value>(), "size of value too big");
	}

	#[rstest]
	// Rev-2 Value envelope overhead: every varlen variant pays 1 tag byte
	// + 4 bytes `u32_le` payload length on top of the legacy `u16 rev` +
	// `u32 disc`. Inline variants (`None`, `Null`) are the entire encoding
	// in a single tag byte. `Bool` is `fixed(1)` -> tag + 1 byte.
	#[case::none(Value::None, 2)]
	#[case::null(Value::Null, 2)]
	#[case::bool(Value::Bool(true), 3)]
	#[case::bool(Value::Bool(false), 3)]
	// String wire shape under rev-2 varlen: rev (1) + tag (1) + u32_le
	// payload_length (4) + Strand wire (`usize len || utf8`) = 6 + Strand.
	// "test" is 4 utf8 bytes + 1 length varint = 5 bytes, so 11 total.
	#[case::string(Value::from("test"), 11)]
	// Objects under rev-2 Value + rev-2 Object `indexed_map`:
	//   Value envelope:  rev=2 (1) + tag (1) + u32_le payload_length (4) = 6
	//   Object envelope: rev=2 (1) + u32_le payload_length (4) = 5
	//   indexed_map body (sub-threshold): flag (1) + varint len + (K, V)*
	// Total delta vs rev-1 Value + rev-1 Object: +13 bytes per Object Value
	// (Value envelope +6, Object envelope +5, indexed_map flag +1, minus
	// 4 bytes the rev-1 framing was paying). The Wire-repr zero-copy
	// `walk_field_0` removed the need for a single-field offset table.
	#[case::object(Value::from(syn::value("{ hello: 'world' }").unwrap()), 31)]
	#[case::object(Value::from(syn::value("{ compact: true, schema: 0 }").unwrap()), 40)]
	fn check_serialize(#[case] value: Value, #[case] expected: usize) {
		let enc: Vec<u8> = revision::to_vec(&value).unwrap();
		assert_eq!(expected, enc.len());
	}

	#[test]
	fn serialize_deserialize() {
		let val = parse_val!(
			"{ test: { something: [1, 'two', null, test:tobie, { trueee: false, noneee: null }] } }"
		);
		let res = parse_val!(
			"{ test: { something: [1, 'two', null, test:tobie, { trueee: false, noneee: null }] } }"
		);
		let enc: Vec<u8> = revision::to_vec(&val).unwrap();
		let dec: Value = revision::from_slice(&enc).unwrap();
		assert_eq!(res, dec);
	}

	#[rstest]
	#[case::none(PublicValue::None, json!(null), PublicValue::Null)]
	#[case::null(PublicValue::Null, json!(null), PublicValue::Null)]
	#[case::bool(PublicValue::Bool(true), json!(true), PublicValue::Bool(true))]
	#[case::bool(PublicValue::Bool(false), json!(false), PublicValue::Bool(false))]
	#[case::number(
		PublicValue::Number(PublicNumber::Int(i64::MIN)),
		json!(i64::MIN),
		PublicValue::Number(PublicNumber::Int(i64::MIN)),
	)]
	#[case::number(
		PublicValue::Number(PublicNumber::Int(i64::MAX)),
		json!(i64::MAX),
		PublicValue::Number(PublicNumber::Int(i64::MAX)),
	)]
	#[case::number(
		PublicValue::Number(PublicNumber::Float(1.23)),
		json!(1.23),
		PublicValue::Number(PublicNumber::Float(1.23)),
	)]
	#[case::number(
		PublicValue::Number(PublicNumber::Float(f64::NEG_INFINITY)),
		json!(null),
		PublicValue::Null,
	)]
	#[case::number(
		PublicValue::Number(PublicNumber::Float(f64::MIN)),
		json!(-1.7976931348623157e308),
		PublicValue::Number(PublicNumber::Float(f64::MIN)),
	)]
	#[case::number(
		PublicValue::Number(PublicNumber::Float(0.0)),
		json!(0.0),
		PublicValue::Number(PublicNumber::Float(0.0)),
	)]
	#[case::number(
		PublicValue::Number(PublicNumber::Float(f64::MAX)),
		json!(1.7976931348623157e308),
		PublicValue::Number(PublicNumber::Float(f64::MAX)),
	)]
	#[case::number(
		PublicValue::Number(PublicNumber::Float(f64::INFINITY)),
		json!(null),
		PublicValue::Null,
	)]
	#[case::number(
		PublicValue::Number(PublicNumber::Float(f64::NAN)),
		json!(null),
		PublicValue::Null,
	)]
	#[case::number(
		PublicValue::Number(PublicNumber::Decimal(Decimal::new(123, 2))),
		json!("1.23"),
		PublicValue::String("1.23".to_string()),
	)]
	#[case::strand(
		PublicValue::String("".to_string()),
		json!(""),
		PublicValue::String("".to_string()),
	)]
	#[case::strand(
		PublicValue::String("foo".to_string()),
		json!("foo"),
		PublicValue::String("foo".to_string()),
	)]
	#[case::duration(
		PublicValue::Duration(PublicDuration::ZERO),
		json!("0ns"),
		PublicValue::String("0ns".to_string()),
	)]
	#[case::duration(
		PublicValue::Duration(PublicDuration::MAX),
		json!("584942417355y3w5d7h15s999ms999µs999ns"),
		PublicValue::String("584942417355y3w5d7h15s999ms999µs999ns".to_string()),
	)]
	#[case::datetime(
		PublicValue::Datetime(PublicDatetime::MIN_UTC),
		json!("-262143-01-01T00:00:00Z"),
		PublicValue::Datetime(PublicDatetime::MIN_UTC),
	)]
	#[case::datetime(
		PublicValue::Datetime(PublicDatetime::MAX_UTC),
		json!("+262142-12-31T23:59:59.999999999Z"),
		PublicValue::Datetime(PublicDatetime::MAX_UTC),
	)]
	#[case::uuid(
		PublicValue::Uuid(PublicUuid::nil()),
		json!("00000000-0000-0000-0000-000000000000"),
		PublicValue::Uuid(PublicUuid::nil()),
	)]
	#[case::uuid(
		PublicValue::Uuid(PublicUuid::max()),
		json!("ffffffff-ffff-ffff-ffff-ffffffffffff"),
		PublicValue::Uuid(PublicUuid::max()),
	)]
	#[case::bytes(
		PublicValue::Bytes(PublicBytes::default()),
		json!([]),
		PublicValue::Array(PublicArray::new()),
	)]
	#[case::bytes(
		PublicValue::Bytes(PublicBytes::from(b"foo".to_vec())),
		json!([102, 111, 111]),
		PublicValue::Array(PublicArray::from(vec![
			PublicValue::Number(PublicNumber::Int(102)),
			PublicValue::Number(PublicNumber::Int(111)),
			PublicValue::Number(PublicNumber::Int(111)),
		])),
	)]
	#[case::record_id(
		PublicValue::RecordId(PublicRecordId::new("foo", PublicRecordIdKey::String("bar".to_string()))) ,
		json!("foo:bar"),
		PublicValue::RecordId(PublicRecordId::new("foo", PublicRecordIdKey::String("bar".to_string()))) ,
	)]
	#[case::array(
		PublicValue::Array(PublicArray::new()),
		json!([]),
		PublicValue::Array(PublicArray::new()),
	)]
	#[case::array(
		PublicValue::Array(PublicArray::from(vec![PublicValue::Bool(true), PublicValue::Bool(false)])),
		json!([true, false]),
		PublicValue::Array(PublicArray::from(vec![PublicValue::Bool(true), PublicValue::Bool(false)])),
	)]
	#[case::object(
		PublicValue::Object(PublicObject::new()),
		json!({}),
		PublicValue::Object(PublicObject::new()),
	)]
	#[case::object(
		PublicValue::Object(PublicObject::from_iter([("done".to_owned(), PublicValue::Bool(true))])),
		json!({"done": true}),
		PublicValue::Object(PublicObject::from_iter([("done".to_owned(), PublicValue::Bool(true))])),
	)]
	#[case::geometry_point(
		PublicValue::Geometry(PublicGeometry::Point(point! { x: 10., y: 20. })),
		json!({ "type": "Point", "coordinates": [10., 20.]}),
		PublicValue::Geometry(PublicGeometry::Point(point! { x: 10., y: 20. })),
	)]
	#[case::geometry_line(
		PublicValue::Geometry(PublicGeometry::Line(line_string![
			( x: 0., y: 0. ),
			( x: 10., y: 0. ),
		])),
		json!({ "type": "LineString", "coordinates": [[0., 0.], [10., 0.]]}),
		PublicValue::Geometry(PublicGeometry::Line(line_string![
			( x: 0., y: 0. ),
			( x: 10., y: 0. ),
		])),
	)]
	#[case::geometry_polygon(
		PublicValue::Geometry(PublicGeometry::Polygon(polygon![
			(x: -111., y: 45.),
			(x: -111., y: 41.),
			(x: -104., y: 41.),
			(x: -104., y: 45.),
		])),
		json!({ "type": "Polygon", "coordinates": [[
			[-111., 45.],
			[-111., 41.],
			[-104., 41.],
			[-104., 45.],
			[-111., 45.],
		]]}),
		PublicValue::Geometry(PublicGeometry::Polygon(polygon![
			(x: -111., y: 45.),
			(x: -111., y: 41.),
			(x: -104., y: 41.),
			(x: -104., y: 45.),
		])),
	)]
	#[case::geometry_multi_point(
		PublicValue::Geometry(PublicGeometry::MultiPoint(MultiPoint::new(vec![
			point! { x: 0., y: 0. },
			point! { x: 1., y: 2. },
		]))),
		json!({ "type": "MultiPoint", "coordinates": [[0., 0.], [1., 2.]]}),
		PublicValue::Geometry(PublicGeometry::MultiPoint(MultiPoint::new(vec![
			point! { x: 0., y: 0. },
			point! { x: 1., y: 2. },
		]))),
	)]
	#[case::geometry_multi_line(
		PublicValue::Geometry(
			PublicGeometry::MultiLine(
				MultiLineString::new(vec![
					line_string![( x: 0., y: 0. ), ( x: 1., y: 2. )],
				])
			)
		),
		json!({ "type": "MultiLineString", "coordinates": [[[0., 0.], [1., 2.]]]}),
		PublicValue::Geometry(
			PublicGeometry::MultiLine(
				MultiLineString::new(vec![
					line_string![( x: 0., y: 0. ), ( x: 1., y: 2. )],
				])
			)
		),
	)]
	#[case::geometry_multi_polygon(
		PublicValue::Geometry(PublicGeometry::MultiPolygon(MultiPolygon::new(vec![
			polygon![
				(x: -111., y: 45.),
				(x: -111., y: 41.),
				(x: -104., y: 41.),
				(x: -104., y: 45.),
			],
		]))),
		json!({ "type": "MultiPolygon", "coordinates": [[[
			[-111., 45.],
			[-111., 41.],
			[-104., 41.],
			[-104., 45.],
			[-111., 45.],
		]]]})
	,	PublicValue::Geometry(PublicGeometry::MultiPolygon(MultiPolygon::new(vec![
			polygon![
				(x: -111., y: 45.),
				(x: -111., y: 41.),
				(x: -104., y: 41.),
				(x: -104., y: 45.),
			],
		]))),
	)]
	#[case::geometry_collection(
		PublicValue::Geometry(PublicGeometry::Collection(vec![])),
		json!({
			"type": "GeometryCollection",
			"geometries": [],
		}),
		PublicValue::Geometry(PublicGeometry::Collection(vec![])),
	)]
	#[case::geometry_collection_with_point(
		PublicValue::Geometry(PublicGeometry::Collection(vec![PublicGeometry::Point(point! { x: 10., y: 20. })])),
		json!({
		"type": "GeometryCollection",
		"geometries": [ { "type": "Point", "coordinates": [10., 20.] } ],
	}),
		PublicValue::Geometry(PublicGeometry::Collection(vec![PublicGeometry::Point(point! { x: 10., y: 20. })])),
	)]
	#[case::geometry_collection_with_line(
		PublicValue::Geometry(PublicGeometry::Collection(vec![PublicGeometry::Line(line_string![
			( x: 0., y: 0. ),
			( x: 10., y: 0. ),
		])])),
		json!({
			"type": "GeometryCollection",
			"geometries": [ { "type": "LineString", "coordinates": [[0., 0.], [10., 0.]] } ],
		}),
		PublicValue::Geometry(PublicGeometry::Collection(vec![PublicGeometry::Line(line_string![
			( x: 0., y: 0. ),
			( x: 10., y: 0. ),
		])])),
	)]

	fn test_json(
		#[case] value: PublicValue,
		#[case] expected: Json,
		#[case] expected_deserialized: PublicValue,
	) {
		use surrealdb_syn::ParserConfig;

		let json_value = value.into_json_value();
		assert_eq!(json_value, expected);

		let json_str = serde_json::to_string(&json_value).expect("Failed to serialize to JSON");
		let deserialized_sql_value =
			crate::syn::value_legacy_strand(&json_str, &ParserConfig::default()).unwrap();
		assert_eq!(deserialized_sql_value, expected_deserialized);
	}
}