1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
//! SourceExpr operator - evaluates an expression and yields values for iteration in FROM clause.
//!
//! This operator is specifically designed for FROM clause sources, handling:
//! - None/Null: yields no rows (empty stream)
//! - Arrays: yields each element as a separate row, resolving RecordIds to full documents
//! - RecordId: fetches the full document and yields it
//! - Other values: yields the value as a single row
//!
//! RecordId resolution ensures downstream pipeline operators (Filter, Sort, etc.)
//! can access document fields, matching the behaviour of RecordIdScan and TableScan.
use std::sync::Arc;
use surrealdb_types::{SqlFormat, ToSql};
use crate::exec::context::{ContextLevel, ExecutionContext};
use crate::exec::physical_expr::{EvalContext, PhysicalExpr};
use crate::exec::{
AccessMode, ExecOperator, FlowResult, OperatorMetrics, ValueBatch, ValueBatchStream,
monitor_stream,
};
use crate::val::Value;
/// SourceExpr operator - evaluates an expression and yields values for iteration.
///
/// Unlike ExprPlan (which returns a single value), SourceExpr handles
/// FROM clause semantics:
/// - None/Null → empty stream (no rows)
/// - Array → yield each element
/// - Other → yield single value
#[derive(Debug, Clone)]
pub struct SourceExpr {
/// The expression to evaluate
pub expr: Arc<dyn PhysicalExpr>,
/// Per-operator runtime metrics for EXPLAIN ANALYZE.
pub(crate) metrics: Arc<OperatorMetrics>,
}
impl SourceExpr {
pub(crate) fn new(expr: Arc<dyn PhysicalExpr>) -> Self {
Self {
expr,
metrics: Arc::new(OperatorMetrics::new()),
}
}
}
impl ExecOperator for SourceExpr {
fn name(&self) -> &'static str {
"SourceExpr"
}
fn attrs(&self) -> Vec<(String, String)> {
vec![("expr".to_string(), self.expr.to_sql())]
}
fn required_context(&self) -> ContextLevel {
// The expression may yield RecordId values that need resolving to full
// documents, which requires database-level context for transaction access.
self.expr.required_context().max(ContextLevel::Database)
}
fn access_mode(&self) -> AccessMode {
// Delegate to the wrapped expression
self.expr.access_mode()
}
fn metrics(&self) -> Option<&OperatorMetrics> {
Some(&self.metrics)
}
fn expressions(&self) -> Vec<(&str, &Arc<dyn PhysicalExpr>)> {
vec![("expr", &self.expr)]
}
fn execute(&self, ctx: &ExecutionContext) -> FlowResult<ValueBatchStream> {
let expr = Arc::clone(&self.expr);
let ctx = ctx.clone();
let stream = async_stream::try_stream! {
// In a correlated subquery, ScalarSubquery binds the outer document
// as `$this`. Use it as current_value so that bare field paths in
// the FROM expression resolve against the outer row (e.g.
// `FROM data.files` where `data` is a record link on the outer row).
let this_value = ctx.value("this").cloned();
let eval_ctx = EvalContext::from_exec_ctx(&ctx);
let eval_ctx = match this_value {
Some(ref v) => eval_ctx.with_value(v),
None => eval_ctx,
};
let value = expr.evaluate(eval_ctx).await?;
match value {
Value::Array(arr) => {
let mut values: Vec<Value> = arr
.into_iter()
.filter(|v| !matches!(v, Value::None | Value::Null))
.collect();
if !values.is_empty() {
// Resolve RecordId values to full documents so that
// downstream operators (Filter, Sort, etc.) can access
// document fields — matching RecordIdScan/TableScan.
super::fetch::batch_fetch_in_place(&ctx, &mut values).await?;
values.retain(|v| !matches!(v, Value::None | Value::Null));
if !values.is_empty() {
yield ValueBatch { values };
}
}
}
Value::None | Value::Null => {}
Value::RecordId(ref rid) => {
let fetched = super::fetch::fetch_record(&ctx, rid).await?;
if !matches!(fetched, Value::None) {
yield ValueBatch { values: vec![fetched] };
}
}
other => {
yield ValueBatch { values: vec![other] };
}
}
};
Ok(monitor_stream(Box::pin(stream), "SourceExpr", &self.metrics))
}
fn is_scalar(&self) -> bool {
// SourceExpr is not scalar - it can yield multiple values
false
}
}
impl ToSql for SourceExpr {
fn fmt_sql(&self, f: &mut String, fmt: SqlFormat) {
self.expr.fmt_sql(f, fmt);
}
}