use surrealdb_types::{SqlFormat, ToSql};
use crate::exec::physical_expr::{EvalContext, PhysicalExpr};
use crate::exec::{AccessMode, BoxFut, ContextLevel};
use crate::expr::FlowResult;
use crate::val::Value;
const PARALLEL_BATCH_THRESHOLD: usize = 2;
#[derive(Debug, Clone)]
pub struct FieldPart {
pub name: String,
}
impl PhysicalExpr for FieldPart {
fn name(&self) -> &'static str {
"Field"
}
fn as_any(&self) -> &dyn std::any::Any {
self
}
fn required_context(&self) -> ContextLevel {
ContextLevel::Database
}
fn evaluate<'a>(&'a self, ctx: EvalContext<'a>) -> BoxFut<'a, FlowResult<Value>> {
Box::pin(async move {
let value = ctx.current_value.unwrap_or(&Value::NONE);
evaluate_field(value, &self.name, ctx).await
})
}
fn evaluate_batch<'a>(
&'a self,
ctx: EvalContext<'a>,
values: &'a [Value],
) -> BoxFut<'a, FlowResult<Vec<Value>>> {
Box::pin(async move {
if values.len() < PARALLEL_BATCH_THRESHOLD {
let mut results = Vec::with_capacity(values.len());
for value in values {
results.push(self.evaluate(ctx.with_value(value)).await?);
}
return Ok(results);
}
let futures: Vec<_> =
values.iter().map(|value| self.evaluate(ctx.with_value(value))).collect();
futures::future::try_join_all(futures).await
})
}
fn access_mode(&self) -> AccessMode {
AccessMode::ReadOnly
}
fn try_simple_field(&self) -> Option<&str> {
Some(&self.name)
}
}
impl ToSql for FieldPart {
fn fmt_sql(&self, f: &mut String, _fmt: SqlFormat) {
f.push('.');
f.push_str(&self.name);
}
}
pub(crate) async fn evaluate_field(
value: &Value,
name: &str,
ctx: EvalContext<'_>,
) -> FlowResult<Value> {
match value {
Value::Object(obj) => Ok(obj.get(name).cloned().unwrap_or(Value::None)),
Value::RecordId(rid) => {
if let crate::val::RecordIdKey::Object(obj) = &rid.key
&& obj.contains_key(name)
{
return Ok(obj.get(name).cloned().unwrap_or(Value::None));
}
if ctx.computing_record.as_ref() == Some(rid) {
let version = ctx.exec_ctx.version_stamp();
let raw =
crate::exec::operators::fetch::fetch_raw_record(ctx.exec_ctx, rid, version)
.await?;
return match raw {
Some(Value::Object(obj)) => Ok(obj.get(name).cloned().unwrap_or(Value::None)),
_ => Ok(Value::None),
};
}
let fetched = if ctx.skip_fetch_perms {
crate::exec::operators::fetch::fetch_record_no_perms(ctx.exec_ctx, rid).await?
} else {
crate::exec::operators::fetch::fetch_record(ctx.exec_ctx, rid).await?
};
match fetched {
Value::Object(obj) => Ok(obj.get(name).cloned().unwrap_or(Value::None)),
_ => Ok(Value::None),
}
}
Value::Array(arr) => {
let mut results = Vec::with_capacity(arr.len());
for v in arr.iter() {
results.push(Box::pin(evaluate_field(v, name, ctx.clone())).await?);
}
Ok(Value::Array(results.into()))
}
Value::Geometry(geo) => {
let obj = geo.as_object();
Ok(obj.get(name).cloned().unwrap_or(Value::None))
}
_ => Ok(Value::None),
}
}