use std::sync::Arc;
use super::Planner;
use crate::err::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::IdiomExpr;
use crate::exec::{ExecOperator, PhysicalExpr};
use crate::expr::part::{DestructurePart as AstDestructurePart, Part, RecurseInstruction};
impl<'ctx> Planner<'ctx> {
pub(crate) async fn convert_idiom(
&self,
idiom: crate::expr::idiom::Idiom,
) -> Result<Arc<dyn PhysicalExpr>, Error> {
use surrealdb_types::ToSql;
let display = idiom.to_sql();
let mut parts = idiom.0;
if let Some(Part::Start(_)) = parts.first() {
let start_part = parts.remove(0);
let Part::Start(start_expr) = start_part else {
return Err(Error::Internal(
"convert_idiom: parts.first() reported Part::Start but parts.remove(0) was not Part::Start"
.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)))
}
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() {
if let Part::Recurse(recurse, None, instruction) = part {
let system_limit = self.ctx.config.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),
};
if min_depth < 1 {
return Err(Error::InvalidBound {
found: min_depth.to_string(),
expected: "at least 1".into(),
});
}
if let Some(max) = max_depth
&& max > system_limit
{
return Err(Error::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);
if instruction.is_some() && has_repeat_recurse {
return Err(Error::RecursionInstructionPlanConflict);
}
let (recurse_body, suffix) = if has_repeat_recurse {
match remaining.iter().position(|p| matches!(p, Part::RepeatRecurse)) {
Some(pos) => {
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>);
if !suffix.is_empty() {
let suffix_parts = self.convert_parts(suffix).await?;
converted.extend(suffix_parts);
}
break;
}
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;
}
if let Part::Lookup(first_lookup) = part {
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(Error::Internal(
"convert_parts lookup fusion: iter.peek() reported Part::Lookup but iter.next() did not yield one"
.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 {
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);
}
let converted = insert_auto_flattens(converted);
Ok(converted)
})
}
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 => {
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(Error::Unreachable(
"Start parts should be handled at the idiom level".to_string(),
)),
Part::Lookup(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.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),
};
if min_depth < 1 {
return Err(Error::InvalidBound {
found: min_depth.to_string(),
expected: "at least 1".into(),
});
}
if let Some(max) = max_depth
&& max > system_limit
{
return Err(Error::InvalidBound {
found: max.to_string(),
expected: format!("{} at most", system_limit),
});
}
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)),
}
}
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;
let start_expr = if matches!(parts.first(), Some(Part::Start(_))) {
let Part::Start(expr) = parts.remove(0) else {
return Err(Error::Internal(
"convert_destructure: parts.first() reported Part::Start but parts.remove(0) was not Part::Start"
.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)
})
}
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,
})
}
}
}
}
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)
}
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,
..
})
)
}
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
}
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
}
}
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
}