use std::collections::HashMap;
use std::sync::Arc;
use surrealdb_strand::Strand;
use surrealdb_types::{SqlFormat, ToSql};
use crate::exec::function::MethodDescriptor;
use crate::exec::physical_expr::function::validate_return;
use crate::exec::physical_expr::{BlockPhysicalExpr, EvalContext, PhysicalExpr};
use crate::exec::{AccessMode, BoxFut, CombineAccessModes, ContextLevel};
use crate::expr::FlowResult;
use crate::val::{Closure, Value};
#[derive(Debug, Clone)]
pub struct MethodPart {
pub descriptor: Arc<MethodDescriptor>,
pub args: Vec<Arc<dyn PhysicalExpr>>,
}
impl PhysicalExpr for MethodPart {
fn name(&self) -> &'static str {
"Method"
}
fn as_any(&self) -> &dyn std::any::Any {
self
}
fn required_context(&self) -> ContextLevel {
self.args.iter().map(|a| a.required_context()).max().unwrap_or(ContextLevel::Root)
}
fn evaluate<'a>(&'a self, ctx: EvalContext<'a>) -> BoxFut<'a, FlowResult<Value>> {
Box::pin(async move {
let value = ctx.current_value.cloned().unwrap_or(Value::None);
let func = self.descriptor.resolve(&value)?;
let mut func_args = Vec::with_capacity(1 + self.args.len());
func_args.push(value);
for arg_expr in &self.args {
let arg_value = arg_expr.evaluate(ctx.clone()).await?;
func_args.push(arg_value);
}
let result = if func.is_pure() && !func.is_async() {
func.invoke(func_args)
} else {
func.invoke_async(&ctx, func_args).await
};
match result {
Ok(v) => Ok(v),
Err(e) => {
if let Some(crate::err::Error::InvalidFunctionArguments {
message,
..
}) = e.downcast_ref::<crate::err::Error>()
{
Err(crate::err::Error::InvalidMethodArguments {
name: self.descriptor.name.to_string(),
message: message.clone(),
}
.into())
} else {
Err(e.into())
}
}
}
})
}
fn access_mode(&self) -> AccessMode {
self.args.iter().map(|a| a.access_mode()).combine_all()
}
}
impl ToSql for MethodPart {
fn fmt_sql(&self, f: &mut String, fmt: SqlFormat) {
f.push('.');
f.push_str(self.descriptor.name);
f.push('(');
for (i, arg) in self.args.iter().enumerate() {
if i > 0 {
f.push_str(", ");
}
arg.fmt_sql(f, fmt);
}
f.push(')');
}
}
#[derive(Debug, Clone)]
pub struct ClosureFieldCallPart {
pub field: String,
pub args: Vec<Arc<dyn PhysicalExpr>>,
}
impl PhysicalExpr for ClosureFieldCallPart {
fn name(&self) -> &'static str {
"ClosureFieldCall"
}
fn as_any(&self) -> &dyn std::any::Any {
self
}
fn required_context(&self) -> ContextLevel {
self.args.iter().map(|a| a.required_context()).max().unwrap_or(ContextLevel::Root)
}
fn evaluate<'a>(&'a self, ctx: EvalContext<'a>) -> BoxFut<'a, FlowResult<Value>> {
Box::pin(async move {
use crate::err::Error;
let value = ctx.current_value.cloned().unwrap_or(Value::None);
let field_value = match &value {
Value::Object(obj) => obj.get(self.field.as_str()).cloned(),
_ => None,
};
let closure = match field_value {
Some(Value::Closure(c)) => c,
_ => {
let type_name = value.kind_of().to_string();
return Err(Error::InvalidFunction {
name: self.field.clone(),
message: format!("no such method found for the {} type", type_name),
}
.into());
}
};
let mut evaluated_args = Vec::with_capacity(self.args.len());
for arg_expr in &self.args {
evaluated_args.push(arg_expr.evaluate(ctx.clone()).await?);
}
match closure.as_ref() {
Closure::Expr {
args: arg_spec,
returns,
body,
captures,
} => {
let mut isolated_ctx = ctx.exec_ctx.clone();
for (name, value) in captures.clone() {
isolated_ctx = isolated_ctx.with_param(name, value);
}
if arg_spec.len() > evaluated_args.len()
&& let Some((param, kind)) =
arg_spec[evaluated_args.len()..].iter().find(|(_, k)| !k.can_be_none())
{
return Err(Error::InvalidFunctionArguments {
name: "ANONYMOUS".to_string(),
message: format!(
"Expected a value of type '{}' for argument {}",
kind.to_sql(),
param.to_sql()
),
}
.into());
}
let mut local_params: HashMap<Strand, Value> = HashMap::new();
for ((param, kind), arg_value) in arg_spec.iter().zip(evaluated_args) {
let coerced = arg_value.coerce_to_kind(kind).map_err(|_| {
Error::InvalidFunctionArguments {
name: "ANONYMOUS".to_string(),
message: format!(
"Expected a value of type '{}' for argument {}",
kind.to_sql(),
param.to_sql()
),
}
})?;
local_params.insert(param.clone().into_strand(), coerced);
}
for (name, value) in &local_params {
isolated_ctx = isolated_ctx.with_param(name.clone(), value.clone());
}
let block_expr = BlockPhysicalExpr {
block: crate::expr::Block(vec![body.clone()]),
};
let eval_ctx = EvalContext {
exec_ctx: &isolated_ctx,
current_value: ctx.current_value,
local_params: Some(&local_params),
recursion_ctx: None,
document_root: ctx.document_root,
skip_fetch_perms: ctx.skip_fetch_perms,
computing_record: ctx.computing_record.clone(),
plan_depth: ctx.plan_depth + 1,
};
let result = match block_expr.evaluate(eval_ctx).await {
Ok(v) => v,
Err(crate::expr::ControlFlow::Return(v)) => v,
Err(crate::expr::ControlFlow::Break)
| Err(crate::expr::ControlFlow::Continue) => {
return Err(Error::InvalidControlFlow.into());
}
Err(e) => return Err(e),
};
Ok(validate_return("ANONYMOUS", returns.as_ref(), result)?)
}
Closure::Builtin(_) => Err(anyhow::anyhow!(
"Builtin closures are not yet supported in the streaming executor"
)
.into()),
}
})
}
fn access_mode(&self) -> AccessMode {
self.args.iter().map(|a| a.access_mode()).combine_all()
}
}
impl ToSql for ClosureFieldCallPart {
fn fmt_sql(&self, f: &mut String, fmt: SqlFormat) {
f.push('.');
f.push_str(&self.field);
f.push('(');
for (i, arg) in self.args.iter().enumerate() {
if i > 0 {
f.push_str(", ");
}
arg.fmt_sql(f, fmt);
}
f.push(')');
}
}