surrealdb-core 3.3.1

//! Idiom and part conversion for the planner.
//!
//! Converts SurrealQL idiom AST nodes to physical part expressions for runtime evaluation.
//! Each part is an `Arc<dyn PhysicalExpr>` that reads its input from `ctx.current_value`.

use std::sync::Arc;

use super::Planner;
use crate::err::{EngineError, Error};
use crate::exec::operators::{CurrentValueSource, RecursionOp};
use crate::exec::parts::{
	AllPart, ClosureFieldCallPart, DestructureField, DestructurePart, FieldPart, FirstPart,
	FlattenPart, IndexPart, LastPart, LookupDirection, LookupPart, MethodPart, OptionalChainPart,
	PhysicalRecurseInstruction, RecursePart, RepeatRecursePart, WherePart,
};
use crate::exec::physical_expr::{FoldedIdiom, IdiomExpr};
use crate::exec::{Error as ExecError, ExecOperator, PhysicalExpr};
use crate::expr::part::{DestructurePart as AstDestructurePart, Part, RecurseInstruction};
use crate::val::Value;

// ============================================================================
// impl Planner — Idiom Conversion
// ============================================================================

impl<'ctx> Planner<'ctx> {
	/// Convert an idiom to a physical expression.
	///
	/// If the first part is `Part::Start(expr)`, the expression is converted to a
	/// physical expression and stored as the start expression on `IdiomExpr`.
	pub(crate) async fn convert_idiom(
		&self,
		idiom: crate::expr::idiom::Idiom,
	) -> Result<Arc<dyn PhysicalExpr>, Error> {
		use surrealdb_types::ToSql;

		// Pre-compute the display string before consuming the idiom's parts
		let display = idiom.to_sql();
		let mut parts = idiom.0;

		// Constant field chains (`$session.data.orgs`) resolve once, here,
		// instead of on every row. See `try_fold_constant_field_chain`.
		//
		// ALTITUDE, so nobody reads more into this than it does: physical-
		// expression construction runs *after* index analysis has chosen the
		// plan. `exec/index/analysis.rs` and `planner/util/conditions.rs` match
		// `Expr::Literal` on the AST and cannot see a `FoldedIdiom`, so
		// `WHERE org IN $session.data.orgs` still full-scans. What this buys is
		// the removal of a per-row deep clone of the whole `$session` object —
		// real, and the reason it is here — not an index and not O(rows): `IN`
		// still walks the payload once per row.
		//
		// The repo's own AST-level pipeline for this is
		// `planner/select/mod.rs`: `resolve_condition_params` ->
		// `fold_condition_expressions` -> `fold_constant_record_idioms`, whose
		// `record_fold.rs` is the object-rooted sibling of this fold, one layer
		// up. The reason a fold had to be bolted in down here is that
		// permission predicates never enter it: `convert_permission_to_physical`
		// (`exec/permission.rs`) goes straight from a catalog `Expr` to a
		// `PhysicalExpr` with no param resolution and no folding, so a
		// permission cannot use an index however it is written, and
		// `plan_idiom` is the only hook it passes through.
		//
		// Running that pipeline over `Permission::Specific`, plus an
		// `Expr::Idiom`-with-constant-start arm in `fold_condition_expressions`,
		// would give this fold, `$auth`/`$token` substitution and index-driven
		// permission checks at once, and would let the `name.as_str() ==
		// "session"` special case in `constant_start_value` become a general
		// rule. That is a restructuring of the permission planning path, not a
		// move of these few lines, and it is deliberately not attempted here.
		if let Some(Part::Start(start)) = parts.first()
			&& parts.len() > 1
			&& let Some(root) = self.constant_start_value(start)
			&& let Some(value) = try_fold_constant_field_chain(&root, &parts[1..])
		{
			return Ok(Arc::new(FoldedIdiom::new(display, value)));
		}

		if let Some(Part::Start(_)) = parts.first() {
			// `peek + remove` instead of one destructure because we need
			// the index check separately. The `else` arm is provably
			// unreachable given the `peek` succeeded — degraded to a typed
			// `EngineError::Internal` so a future Part::Start variant change
			// surfaces a useful planner-bug message rather than a panic on
			// production user input.
			let start_part = parts.remove(0);
			let Part::Start(start_expr) = start_part else {
				return Err(EngineError::Internal(
					"convert_idiom: parts.first() reported Part::Start but parts.remove(0) was not Part::Start"
						.into(),
				).into());
			};
			let start_phys = self.physical_expr(start_expr).await?;
			let remaining_parts = self.convert_parts(parts).await?;
			return Ok(Arc::new(IdiomExpr::new(display, Some(start_phys), remaining_parts)));
		}

		let physical_parts = self.convert_parts(parts).await?;
		Ok(Arc::new(IdiomExpr::new(display, None, physical_parts)))
	}

	/// Convert idiom parts to physical part expressions.
	///
	/// When a `Part::Recurse` has no explicit inner path, all remaining parts
	/// after the recurse are absorbed as the recursion's inner path.
	///
	/// This function also handles two planner-time responsibilities:
	/// 1. Inserts `FlattenPart` between consecutive `LookupPart`/`WherePart` pairs.
	/// 2. Wraps remaining parts in `OptionalChainPart` when encountering `Part::Optional`.
	pub(crate) fn convert_parts(
		&self,
		parts: Vec<Part>,
	) -> crate::exec::BoxFut<'_, Result<Vec<Arc<dyn PhysicalExpr>>, Error>> {
		Box::pin(async move {
			let mut converted = Vec::with_capacity(parts.len());
			let mut iter = parts.into_iter().peekable();

			while let Some(part) = iter.next() {
				// Handle implicit recursion (Recurse with no inner path absorbs remaining parts)
				if let Part::Recurse(recurse, None, instruction) = part {
					let system_limit = self.ctx.config.exec.idiom_recursion_limit;
					let (min_depth, max_depth) = match recurse {
						crate::expr::part::Recurse::Fixed(n) => (n, Some(n)),
						crate::expr::part::Recurse::Range(min, max) => (min.unwrap_or(1), max),
					};

					// Validate bounds (same checks as expr/part.rs)
					if min_depth < 1 {
						return Err(Error::Exec(ExecError::InvalidBound {
							found: min_depth.to_string(),
							expected: "at least 1".into(),
						}));
					}
					if let Some(max) = max_depth
						&& max > system_limit
					{
						return Err(Error::Exec(ExecError::InvalidBound {
							found: max.to_string(),
							expected: format!("{} at most", system_limit),
						}));
					}

					let remaining: Vec<Part> = iter.collect();
					let has_repeat_recurse = ast_contains_repeat_recurse(&remaining);

					// When an instruction is provided (e.g. +path, +collect)
					// AND the path contains a repeat recurse (@), the two
					// conflict -- match the old compute path's check.
					if instruction.is_some() && has_repeat_recurse {
						return Err(Error::Exec(ExecError::RecursionInstructionPlanConflict));
					}

					// When the path contains @, split at the @ marker.
					// Parts up to and including @ form the recursion body.
					// Parts after @ are a suffix applied once after recursion
					// completes -- matching the old compute path's
					// split_by_repeat_recurse semantics.
					let (recurse_body, suffix) = if has_repeat_recurse {
						match remaining.iter().position(|p| matches!(p, Part::RepeatRecurse)) {
							Some(pos) => {
								// Include everything up to and including @
								let body = remaining[..=pos].to_vec();
								let after = remaining[pos + 1..].to_vec();
								(body, after)
							}
							None => (remaining, vec![]),
						}
					} else {
						(remaining, vec![])
					};

					let path = self.convert_parts(recurse_body).await?;
					let inclusive = is_inclusive_recurse(&instruction);
					let instr = self.convert_recurse_instruction(instruction).await?;

					let body_op = extract_body_operator(&path);
					let op: Arc<dyn ExecOperator> = Arc::new(RecursionOp::new(
						body_op,
						path,
						min_depth,
						max_depth,
						instr,
						inclusive,
						has_repeat_recurse,
					));
					converted.push(Arc::new(RecursePart {
						op,
					}) as Arc<dyn PhysicalExpr>);

					// Append suffix parts (parts after @) outside the
					// recursion so they are evaluated once on the final
					// result rather than at every recursion depth.
					if !suffix.is_empty() {
						let suffix_parts = self.convert_parts(suffix).await?;
						converted.extend(suffix_parts);
					}

					break;
				}

				// Handle optional chaining -- wrap all remaining parts in OptionalChainPart
				if matches!(part, Part::Optional) {
					let remaining: Vec<Part> = iter.collect();
					let tail = self.convert_parts(remaining).await?;
					converted.push(Arc::new(OptionalChainPart {
						tail,
					}) as Arc<dyn PhysicalExpr>);
					break;
				}

				// Fuse consecutive Lookup parts into a single operator chain.
				// Instead of creating independent LookupParts each with their own
				// CurrentValueSource, we thread the output of one lookup as the
				// input to the next, forming a streaming DAG:
				//   GraphEdgeScan(->person, GraphEdgeScan(->likes, CurrentValueSource))
				//
				// When a consecutive `->edge->vertex` pair is fast-path
				// eligible (no edge-side filters/projections, edge table has
				// `PERMISSIONS FULL`), both hops collapse into a single scan
				// in `TargetVertex` mode that reads the target vertex from
				// the adjacency key directly. See [`try_fast_path_pair`].
				if let Part::Lookup(first_lookup) = part {
					// Collect every consecutive lookup so we can inspect pairs.
					// `Part::Lookup` boxes its payload (#209), so we deref each
					// one as we collect.
					let mut lookups: Vec<crate::expr::lookup::Lookup> = vec![*first_lookup];
					while matches!(iter.peek(), Some(Part::Lookup(_))) {
						let Some(Part::Lookup(lu)) = iter.next() else {
							return Err(EngineError::Internal(
								"convert_parts lookup fusion: iter.peek() reported Part::Lookup but iter.next() did not yield one"
									.into(),
							).into());
						};
						lookups.push(*lu);
					}
					let fused = lookups.len() > 1;
					let only = lookups.iter().any(|l| l.only);

					let mut chain: Arc<dyn ExecOperator> = Arc::new(CurrentValueSource::new());
					let mut direction = LookupDirection::Out;
					let mut extract_id = false;

					let mut idx = 0;
					while idx < lookups.len() {
						let advance = if idx + 1 < lookups.len() {
							self.try_fast_path_pair(&lookups[idx], &lookups[idx + 1]).await?
						} else {
							None
						};
						if let Some(plan) = advance {
							// Fast path: replace the (edge, vertex) pair with
							// a single TargetVertex scan whose metadata is
							// taken from the *vertex* lookup so downstream
							// behavior matches the original two-scan chain.
							chain = self.plan_target_vertex_scan(chain, plan).await?;
							let (d, e) = lookup_metadata(&lookups[idx + 1]);
							direction = d;
							extract_id = e;
							idx += 2;
						} else {
							let lu = lookups[idx].clone();
							let (d, e) = lookup_metadata(&lu);
							direction = d;
							extract_id = e;
							chain = self.plan_lookup_with_input(chain, lu).await?;
							idx += 1;
						}
					}

					converted.push(Arc::new(LookupPart {
						direction,
						plan: chain,
						extract_id,
						fused,
						only,
					}));
					continue;
				}

				let physical_part = self.convert_part(part).await?;
				converted.push(physical_part);
			}

			// Insert FlattenPart between consecutive Lookup and Lookup/Where pairs.
			// This is a planner-time responsibility that replaces the runtime introspection
			// that was previously done in IdiomExpr's evaluation loop.
			let converted = insert_auto_flattens(converted);

			Ok(converted)
		})
	}

	/// Convert a single `Part` to an `Arc<dyn PhysicalExpr>`.
	pub(crate) async fn convert_part(&self, part: Part) -> Result<Arc<dyn PhysicalExpr>, Error> {
		match part {
			Part::Field(name) => Ok(Arc::new(FieldPart {
				name: name.into_string(),
			})),

			Part::Value(expr) => {
				let phys_expr = self.physical_expr(expr).await?;
				Ok(Arc::new(IndexPart {
					expr: phys_expr,
				}))
			}

			Part::All => Ok(Arc::new(AllPart)),
			Part::Flatten => Ok(Arc::new(FlattenPart)),
			Part::First => Ok(Arc::new(FirstPart)),
			Part::Last => Ok(Arc::new(LastPart)),

			Part::Optional => {
				// Standalone optional without remaining parts -- should not happen
				// normally because convert_parts handles it, but handle gracefully.
				Ok(Arc::new(OptionalChainPart {
					tail: vec![],
				}))
			}

			Part::Where(expr) => {
				let needs_parent = super::row_scope::references_parent(&expr);
				let phys_expr = self.physical_expr(expr).await?;
				Ok(Arc::new(WherePart {
					predicate: phys_expr,
					needs_parent,
				}))
			}

			Part::Method(name, args) => {
				let mut phys_args = Vec::with_capacity(args.len());
				for arg in args {
					phys_args.push(self.physical_expr(arg).await?);
				}
				let registry = self.function_registry();
				match registry.get_method(name.as_str()) {
					Some(descriptor) => Ok(Arc::new(MethodPart {
						descriptor: Arc::clone(descriptor),
						args: phys_args,
					})),
					None => Ok(Arc::new(ClosureFieldCallPart {
						field: name.into_string(),
						args: phys_args,
					})),
				}
			}

			Part::Destructure(parts) => {
				let fields = self.convert_destructure(parts).await?;
				Ok(Arc::new(DestructurePart {
					fields,
				}))
			}

			Part::Start(_) => Err(EngineError::Unreachable(
				"Start parts should be handled at the idiom level".to_string(),
			)
			.into()),

			Part::Lookup(lookup) => {
				// Extract metadata before consuming lookup
				let direction = match &lookup.kind {
					crate::expr::lookup::LookupKind::Graph(dir) => LookupDirection::from(dir),
					crate::expr::lookup::LookupKind::Reference => LookupDirection::Reference,
				};
				let needs_full_pipeline = lookup.expr.is_some() || lookup.group.is_some();
				let needs_full_records =
					needs_full_pipeline || lookup.cond.is_some() || lookup.split.is_some();
				let extract_id = needs_full_records && !needs_full_pipeline;
				let only = lookup.only;
				let plan = self.plan_lookup(*lookup).await?;
				Ok(Arc::new(LookupPart {
					direction,
					plan,
					extract_id,
					fused: false,
					only,
				}))
			}

			Part::Recurse(recurse, inner_path, instruction) => {
				let system_limit = self.ctx.config.exec.idiom_recursion_limit;
				let (min_depth, max_depth) = match recurse {
					crate::expr::part::Recurse::Fixed(n) => (n, Some(n)),
					crate::expr::part::Recurse::Range(min, max) => (min.unwrap_or(1), max),
				};

				// Validate bounds (same checks as expr/part.rs)
				if min_depth < 1 {
					return Err(ExecError::InvalidBound {
						found: min_depth.to_string(),
						expected: "at least 1".into(),
					}
					.into());
				}
				if let Some(max) = max_depth
					&& max > system_limit
				{
					return Err(ExecError::InvalidBound {
						found: max.to_string(),
						expected: format!("{} at most", system_limit),
					}
					.into());
				}

				let (path, has_repeat_recurse) = if let Some(p) = inner_path {
					let has_rr = ast_contains_repeat_recurse(&p.0);
					let converted = self.convert_parts(p.0).await?;
					(converted, has_rr)
				} else {
					(vec![], false)
				};

				let inclusive = is_inclusive_recurse(&instruction);
				let instr = self.convert_recurse_instruction(instruction).await?;

				let body_op = extract_body_operator(&path);
				let op: Arc<dyn ExecOperator> = Arc::new(RecursionOp::new(
					body_op,
					path,
					min_depth,
					max_depth,
					instr,
					inclusive,
					has_repeat_recurse,
				));
				Ok(Arc::new(RecursePart {
					op,
				}))
			}

			Part::Doc => Ok(Arc::new(FieldPart {
				name: "id".to_string(),
			})),

			Part::RepeatRecurse => Ok(Arc::new(RepeatRecursePart)),
		}
	}

	/// Convert destructure parts to physical destructure fields.
	pub(crate) fn convert_destructure(
		&self,
		parts: Vec<AstDestructurePart>,
	) -> crate::exec::BoxFut<'_, Result<Vec<DestructureField>, Error>> {
		Box::pin(async move {
			let mut fields = Vec::with_capacity(parts.len());

			for part in parts {
				let field = match part {
					AstDestructurePart::All(name) => DestructureField::All(name),
					AstDestructurePart::Field(name) => DestructureField::Field(name),
					AstDestructurePart::Aliased(name, idiom) => {
						let mut parts = idiom.0;
						// Handle Part::Start the same way convert_idiom does:
						// extract it, convert to a physical expression, and prepend
						// to the path. This occurs when the parser wraps non-idiom
						// expressions (like $this or level * 2) in Part::Start
						// within destructure aliases. The `else` arm is provably
						// unreachable; converted to `EngineError::Internal` for the
						// same reason as the matching site in `convert_idiom`.
						let start_expr = if matches!(parts.first(), Some(Part::Start(_))) {
							let Part::Start(expr) = parts.remove(0) else {
								return Err(EngineError::Internal(
									"convert_destructure: parts.first() reported Part::Start but parts.remove(0) was not Part::Start"
										.into(),
								)
								.into());
							};
							Some(self.physical_expr(expr).await?)
						} else {
							None
						};
						let mut path = self.convert_parts(parts).await?;
						if let Some(start) = start_expr {
							path.insert(0, start);
						}
						DestructureField::Aliased {
							field: name,
							path,
						}
					}
					AstDestructurePart::Destructure(name, nested) => {
						let nested_fields = self.convert_destructure(nested).await?;
						DestructureField::Nested {
							field: name,
							parts: nested_fields,
						}
					}
				};
				fields.push(field);
			}

			Ok(fields)
		})
	}

	/// Convert a `RecurseInstruction` to a `PhysicalRecurseInstruction`.
	pub(crate) async fn convert_recurse_instruction(
		&self,
		instruction: Option<RecurseInstruction>,
	) -> Result<PhysicalRecurseInstruction, Error> {
		match instruction {
			None => Ok(PhysicalRecurseInstruction::Default),
			Some(RecurseInstruction::Collect {
				..
			}) => Ok(PhysicalRecurseInstruction::Collect),
			Some(RecurseInstruction::Path {
				..
			}) => Ok(PhysicalRecurseInstruction::Path),
			Some(RecurseInstruction::Shortest {
				expects,
				..
			}) => {
				let target = self.physical_expr(expects).await?;
				Ok(PhysicalRecurseInstruction::Shortest {
					target,
				})
			}
		}
	}
}

// ============================================================================
// Free Functions
// ============================================================================

/// Extract direction and extract_id metadata from a Lookup AST node.
///
/// Used by `convert_parts` to compute `LookupPart` metadata when fusing
/// consecutive lookups. The last lookup's metadata is used for the
/// resulting `LookupPart`.
fn lookup_metadata(lookup: &crate::expr::lookup::Lookup) -> (LookupDirection, bool) {
	let direction = match &lookup.kind {
		crate::expr::lookup::LookupKind::Graph(dir) => LookupDirection::from(dir),
		crate::expr::lookup::LookupKind::Reference => LookupDirection::Reference,
	};
	let needs_full_pipeline = lookup.expr.is_some() || lookup.group.is_some();
	let needs_full_records = needs_full_pipeline || lookup.cond.is_some() || lookup.split.is_some();
	let extract_id = needs_full_records && !needs_full_pipeline;
	(direction, extract_id)
}

/// Check if a `RecurseInstruction` has the `inclusive` flag set.
fn is_inclusive_recurse(instruction: &Option<RecurseInstruction>) -> bool {
	matches!(
		instruction,
		Some(RecurseInstruction::Path {
			inclusive: true,
			..
		}) | Some(RecurseInstruction::Collect {
			inclusive: true,
			..
		}) | Some(RecurseInstruction::Shortest {
			inclusive: true,
			..
		})
	)
}

/// Insert `FlattenPart` between consecutive `LookupPart` and `LookupPart`/`WherePart` pairs.
///
/// This is a planner-time transformation that replaces the runtime introspection
/// that was previously done in `IdiomExpr`'s evaluation loop and `evaluate_physical_path`.
fn insert_auto_flattens(parts: Vec<Arc<dyn PhysicalExpr>>) -> Vec<Arc<dyn PhysicalExpr>> {
	if parts.len() < 2 {
		return parts;
	}

	let mut result = Vec::with_capacity(parts.len() * 2);
	for i in 0..parts.len() {
		result.push(Arc::clone(&parts[i]));
		if parts[i].name() == "Lookup"
			&& let Some(next) = parts.get(i + 1)
			&& (next.name() == "Lookup" || next.name() == "Where")
		{
			result.push(Arc::new(FlattenPart) as Arc<dyn PhysicalExpr>);
		}
	}
	result
}

/// Extract operator chain from body parts for EXPLAIN display.
///
/// Collects embedded operators from all path parts (LookupParts expose
/// their fused chains). Returns the chain if exactly one is found.
fn extract_body_operator(path: &[Arc<dyn PhysicalExpr>]) -> Option<Arc<dyn ExecOperator>> {
	let embedded: Vec<_> =
		path.iter().flat_map(|p| p.embedded_operators()).map(|(_, op)| Arc::clone(op)).collect();
	if embedded.len() == 1 {
		Some(embedded.into_iter().next().expect("embedded operator should be present"))
	} else {
		None
	}
}

/// Check if a slice of AST parts contains a `RepeatRecurse` marker at any nesting level.
///
/// This operates on the AST representation (before conversion to PhysicalExpr)
/// so it has full access to the part structure without needing downcasting.
fn ast_contains_repeat_recurse(parts: &[Part]) -> bool {
	for part in parts {
		match part {
			Part::RepeatRecurse => return true,
			Part::Destructure(dest_parts) => {
				for dp in dest_parts {
					if let AstDestructurePart::Aliased(_, idiom) = dp
						&& ast_contains_repeat_recurse(&idiom.0)
					{
						return true;
					}
					if let AstDestructurePart::Destructure(_, nested) = dp {
						for np in nested {
							if let AstDestructurePart::Aliased(_, idiom) = np
								&& ast_contains_repeat_recurse(&idiom.0)
							{
								return true;
							}
						}
					}
				}
			}
			Part::Recurse(_, Some(inner_path), _) if ast_contains_repeat_recurse(&inner_path.0) => {
				return true;
			}
			_ => {}
		}
	}
	false
}

impl Planner<'_> {
	/// The constant value an idiom's start denotes, when it denotes one.
	///
	/// Two shapes qualify, and only two:
	///
	/// * `Expr::Literal` — already a constant in the tree, so reading it needs no environment at
	///   all. This is the shape a `WHERE` clause has by the time it reaches here, because
	///   `resolve_condition_params` has already substituted its parameters for literals.
	/// * `$session` — read out of the very context this plan will run in, so the value folded in is
	///   the one the evaluator would have read. `LET`, `UNSET`, `db.set` and RPC `set` all refuse
	///   the name (it is in `PROTECTED_PARAM_NAMES`), so no *statement* can rebind it mid-plan.
	///
	/// Every other parameter is left alone. A `DEFINE PARAM` carries permissions
	/// that belong at runtime, and an ordinary binding may be rebound by an
	/// enclosing construct after the plan is built.
	///
	/// SECURITY: `$session` is **not** unshadowable, and nothing here should be
	/// read as claiming it is. `PROTECTED_PARAM_NAMES` is consulted only by the
	/// statement forms listed above; the constructs that bind a user-chosen
	/// parameter name — `FOR $session IN [...]`, closure parameters, user-defined
	/// function arguments — bind it unchecked, and `Context::value` prefers a
	/// local insert over the parent chain. The same routes shadow `$auth`, which
	/// every record permission already depends on, so folding `$session` here
	/// neither adds nor removes trust: it inherits exactly `$auth`'s boundary.
	///
	/// SECURITY: folding bakes one session's values into a physical plan. That is
	/// sound only because plans are never shared between sessions — there is no
	/// cross-session plan cache in the engine today, and a plan built under one
	/// binding of `$session` is never reused under another. If such a cache is
	/// ever added it must either key on the folded session values or exclude
	/// plans that carry them, or this turns into a cross-tenant leak.
	fn constant_start_value(&self, start: &crate::expr::Expr) -> Option<Value> {
		use crate::expr::Expr;
		match start {
			Expr::Literal(_) => start.as_static_value(),
			// Resolved from the ambient planning context. `None` when no
			// session is attached, in which case nothing folds.
			Expr::Param(name) if name.as_str() == "session" => self.ctx.value("session").cloned(),
			_ => None,
		}
	}
}

/// Walk a chain of plain field accesses over constant objects.
///
/// Deliberately narrow: every part must be a `Part::Field`, and every value it
/// steps through must be a `Value::Object`. Requiring an object at each step is
/// what preserves record-link semantics — `$auth.email` starts at a
/// `Value::RecordId`, whose field access is a *fetch of the stored row at
/// runtime*, so it never folds. It also keeps the fold away from every part
/// kind that has runtime behaviour of its own (`[WHERE ...]`, `.*`, method
/// calls, recursion, destructuring).
///
/// A missing key folds to `NONE`, which is what the field access evaluates to.
///
/// Deliberately *not* guarded by `substitution_is_lossy`. That predicate exists
/// for `ParamResolver`, which rewrites the AST (`*expr = value.into_literal()`)
/// — and `Value::Set`'s literal form is an array literal, which evaluates to an
/// array and compares unequal to a set. Nothing here round-trips through SQL:
/// `FoldedIdiom::evaluate` hands back `self.value.clone()` verbatim and
/// `try_literal` hands out a `&Value` that `SimpleBinaryOp` and the pre-decode
/// filter consume as a `Value`, through the same `operate::*` functions the
/// unfolded `BinaryOp` would have used. A set folds correctly, and refusing one
/// only meant that `CONTEXT (SELECT VALUE org FROM ...)` returning a set — or
/// any object payload holding one at any depth — fell back to cloning the whole
/// `$session` object per row, which is the exact cost this fold exists to
/// remove.
fn try_fold_constant_field_chain(root: &Value, parts: &[Part]) -> Option<Value> {
	let mut current = root;
	for (i, part) in parts.iter().enumerate() {
		let Part::Field(name) = part else {
			return None;
		};
		let Value::Object(object) = current else {
			return None;
		};
		match object.get(name.as_str()) {
			Some(next) => current = next,
			// A missing key evaluates to NONE, but only the last step can be
			// folded that way. Anything after it operates on NONE and has its
			// own behaviour there — `{}.prop.is_bool()` is `false`, not NONE —
			// so the rest is left to the evaluator.
			None if i + 1 == parts.len() => return Some(Value::None),
			None => return None,
		}
	}
	Some(current.clone())
}

#[cfg(test)]
mod fold_tests {
	use super::*;
	use crate::val::{Object, RecordId, Value};

	fn field(name: &str) -> Part {
		Part::Field(surrealdb_strand::Strand::new(name))
	}

	fn object(pairs: Vec<(&str, Value)>) -> Value {
		let mut obj = Object::default();
		for (k, v) in pairs {
			obj.insert(k, v);
		}
		Value::Object(obj)
	}

	#[test]
	fn folds_a_chain_of_object_fields() {
		let root = object(vec![("data", object(vec![("org", Value::from("acme"))]))]);
		assert_eq!(
			try_fold_constant_field_chain(&root, &[field("data"), field("org")]),
			Some(Value::from("acme"))
		);
	}

	#[test]
	fn a_missing_key_folds_to_none_only_as_the_last_step() {
		let root = object(vec![("data", Value::None)]);
		assert_eq!(try_fold_constant_field_chain(&root, &[field("absent")]), Some(Value::None));
		// `{}.prop.is_bool()` is `false`, not NONE: once a step is missing,
		// whatever follows has its own behaviour on NONE and must be left to
		// the evaluator.
		assert_eq!(try_fold_constant_field_chain(&root, &[field("absent"), field("deeper")]), None);
	}

	/// The stopping rule that preserves record-link semantics: `$auth.email`
	/// must stay a runtime fetch of the stored row, never a plan-time pick.
	#[test]
	fn refuses_to_step_through_a_record_link() {
		let root = object(vec![(
			"rd",
			Value::RecordId(RecordId::new(crate::val::TableName::new("user"), 1)),
		)]);
		assert_eq!(try_fold_constant_field_chain(&root, &[field("rd"), field("email")]), None);
		// And a record id as the whole root is refused for the same reason.
		let rid = Value::RecordId(RecordId::new(crate::val::TableName::new("user"), 1));
		assert_eq!(try_fold_constant_field_chain(&rid, &[field("email")]), None);
	}

	/// Anything with runtime behaviour of its own stops the fold.
	#[test]
	fn refuses_non_field_parts() {
		let root = object(vec![("data", Value::from("acme"))]);
		assert_eq!(try_fold_constant_field_chain(&root, &[Part::All]), None);
		assert_eq!(try_fold_constant_field_chain(&root, &[field("data"), Part::First]), None);
	}

	/// A set folds, and folds as a set.
	///
	/// `ParamResolver` refuses one because it substitutes a *literal* into the
	/// AST and `Value::Set`'s literal form is an array literal. Nothing here
	/// goes through SQL text, so the set survives — and it has to, because a
	/// `CONTEXT` clause returning a set is exactly the payload shape this fold
	/// exists for. Folding it to an array instead would be the bug.
	#[test]
	fn folds_a_set_without_flattening_it_to_an_array() {
		let mut set = crate::val::Set::new();
		set.insert(Value::from("acme"));
		set.insert(Value::from("globex"));
		let folded = try_fold_constant_field_chain(
			&object(vec![("orgs", Value::Set(set.clone()))]),
			&[field("orgs")],
		);
		assert_eq!(folded, Some(Value::Set(set.clone())));
		assert!(!matches!(folded, Some(Value::Array(_))), "a set must not fold to an array");

		// And nested at depth, which is where `substitution_is_lossy`'s
		// recursion into `Object::values` used to disable the fold for a whole
		// payload because of one key.
		let root = object(vec![(
			"data",
			object(vec![("orgs", Value::Set(set.clone())), ("name", Value::from("alice"))]),
		)]);
		assert_eq!(
			try_fold_constant_field_chain(&root, &[field("data"), field("name")]),
			Some(Value::from("alice"))
		);
		assert_eq!(
			try_fold_constant_field_chain(&root, &[field("data")]),
			Some(object(vec![("orgs", Value::Set(set)), ("name", Value::from("alice"))]))
		);
	}
}