use std::fmt::Write;
use std::sync::Arc;
use futures::{StreamExt, stream};
use surrealdb_types::ToSql;
use crate::exec::context::{ContextLevel, ExecutionContext};
use crate::exec::{
AccessMode, CardinalityHint, ExecOperator, FlowResult, OperatorMetrics, ValueBatch,
ValueBatchStream, buffer_stream,
};
use crate::expr::{ControlFlow, ExplainFormat};
use crate::val::{Array, Object, Value};
const INDENT_WIDTH: usize = 4;
#[derive(Debug)]
pub struct ExplainPlan {
pub plan: Arc<dyn ExecOperator>,
pub format: ExplainFormat,
}
impl ExecOperator for ExplainPlan {
fn name(&self) -> &'static str {
"Explain"
}
fn attrs(&self) -> Vec<(String, String)> {
match self.format {
ExplainFormat::Text => vec![("format".to_string(), "TEXT".to_string())],
ExplainFormat::Json => vec![("format".to_string(), "JSON".to_string())],
}
}
fn required_context(&self) -> ContextLevel {
ContextLevel::Root
}
fn access_mode(&self) -> AccessMode {
AccessMode::ReadOnly
}
fn cardinality_hint(&self) -> CardinalityHint {
CardinalityHint::AtMostOne
}
fn execute(&self, _ctx: &ExecutionContext) -> FlowResult<ValueBatchStream> {
let output = match self.format {
ExplainFormat::Text => {
let mut plan_text = String::new();
format_execution_plan(self.plan.as_ref(), &mut plan_text, "");
Value::String(plan_text.into())
}
ExplainFormat::Json => {
let plan_json = format_execution_plan_json(self.plan.as_ref());
Value::Object(plan_json)
}
};
Ok(Box::pin(stream::once(async move {
Ok(ValueBatch {
values: vec![output],
})
})))
}
fn is_scalar(&self) -> bool {
true
}
}
#[derive(Debug)]
pub struct AnalyzePlan {
pub plan: Arc<dyn ExecOperator>,
pub format: ExplainFormat,
pub redact_volatile_explain_attrs: bool,
}
impl ExecOperator for AnalyzePlan {
fn name(&self) -> &'static str {
"ExplainAnalyze"
}
fn attrs(&self) -> Vec<(String, String)> {
match self.format {
ExplainFormat::Text => vec![("format".to_string(), "TEXT".to_string())],
ExplainFormat::Json => vec![("format".to_string(), "JSON".to_string())],
}
}
fn required_context(&self) -> ContextLevel {
self.plan.required_context()
}
fn access_mode(&self) -> AccessMode {
self.plan.access_mode()
}
fn cardinality_hint(&self) -> CardinalityHint {
CardinalityHint::AtMostOne
}
fn children(&self) -> Vec<&Arc<dyn ExecOperator>> {
vec![&self.plan]
}
fn execute(&self, ctx: &ExecutionContext) -> FlowResult<ValueBatchStream> {
self.plan.enable_metrics();
let mut inner_stream = buffer_stream(
self.plan.execute(ctx)?,
self.plan.access_mode(),
self.plan.cardinality_hint(),
ctx.root().ctx.config.operator_buffer_size,
);
let plan = Arc::clone(&self.plan);
let format = self.format;
let redact_volatile_explain_attrs = self.redact_volatile_explain_attrs;
let analyze_stream = async_stream::try_stream! {
let mut total_rows: u64 = 0;
while let Some(batch_result) = inner_stream.next().await {
match batch_result {
Ok(batch) => {
total_rows += batch.values.len() as u64;
}
Err(ControlFlow::Break | ControlFlow::Return(_)) => break,
Err(ControlFlow::Continue) => continue,
Err(e @ ControlFlow::Err(_)) => Err(e)?,
}
}
let output = match format {
ExplainFormat::Text => {
let mut plan_text = String::new();
format_analyze_plan(plan.as_ref(), &mut plan_text, "", redact_volatile_explain_attrs);
let _ = writeln!(plan_text);
let _ = write!(plan_text, "Total rows: {}", total_rows);
Value::String(plan_text.into())
}
ExplainFormat::Json => {
let mut plan_json = format_analyze_plan_json(plan.as_ref(), redact_volatile_explain_attrs);
plan_json.insert("total_rows", Value::from(total_rows as i64));
Value::Object(plan_json)
}
};
yield ValueBatch {
values: vec![output],
};
};
Ok(Box::pin(analyze_stream))
}
fn is_scalar(&self) -> bool {
true
}
}
fn format_execution_plan(plan: &dyn ExecOperator, output: &mut String, prefix: &str) {
let name = plan.name();
let properties = plan.attrs();
let context = plan.required_context();
let _ = write!(output, "{} [ctx: {}]", name, context.short_name());
if !properties.is_empty() {
let _ = write!(output, " [");
for (i, (key, value)) in properties.iter().enumerate() {
if i > 0 {
let _ = write!(output, ", ");
}
let _ = write!(output, "{key}: {value}");
}
let _ = write!(output, "]");
}
let _ = writeln!(output);
let expressions = plan.expressions();
for (role, expr) in &expressions {
let embedded = expr.embedded_operators();
if !embedded.is_empty() {
for (embed_role, embed_plan) in &embedded {
let _ = write!(output, "{} {}.{}: ", prefix, role, embed_role);
format_execution_plan(embed_plan.as_ref(), output, &format!("{} ", prefix));
}
}
}
let children = plan.children();
if !children.is_empty() {
let child_prefix = format!("{}{:width$}", prefix, "", width = INDENT_WIDTH);
for child in children.iter() {
let _ = write!(output, "{}", child_prefix);
format_execution_plan(child.as_ref(), output, &child_prefix);
}
}
}
fn format_execution_plan_json(plan: &dyn ExecOperator) -> Object {
let mut obj = Object::default();
obj.insert("operator", Value::String(plan.name().into()));
obj.insert("context", Value::String(plan.required_context().short_name().into()));
let attrs = plan.attrs();
if !attrs.is_empty() {
let mut attrs_obj = Object::default();
for (key, value) in attrs {
attrs_obj.insert(key, Value::String(value.into()));
}
obj.insert("attributes", Value::Object(attrs_obj));
}
let expressions = plan.expressions();
if !expressions.is_empty() {
let exprs_arr: Vec<Value> = expressions
.iter()
.map(|(role, expr)| {
let mut expr_obj = Object::default();
expr_obj.insert("role", Value::String((*role).into()));
expr_obj.insert("sql", Value::String(expr.to_sql().into()));
let embedded = expr.embedded_operators();
if !embedded.is_empty() {
let embedded_arr: Vec<Value> = embedded
.iter()
.map(|(embed_role, embed_plan)| {
let mut e = Object::default();
e.insert("role", Value::String((*embed_role).into()));
e.insert(
"plan",
Value::Object(format_execution_plan_json(embed_plan.as_ref())),
);
Value::Object(e)
})
.collect();
expr_obj.insert("embedded_operators", Value::Array(Array::from(embedded_arr)));
}
Value::Object(expr_obj)
})
.collect();
obj.insert("expressions", Value::Array(Array::from(exprs_arr)));
}
let children = plan.children();
if !children.is_empty() {
let children_array: Vec<Value> = children
.iter()
.map(|child| Value::Object(format_execution_plan_json(child.as_ref())))
.collect();
obj.insert("children", Value::Array(Array::from(children_array)));
}
obj
}
fn format_metrics_text(metrics: &OperatorMetrics, redact_volatile_explain_attrs: bool) -> String {
let rows = metrics.output_rows();
if redact_volatile_explain_attrs {
return format!("rows: {}", rows);
}
let batches = metrics.output_batches();
let elapsed = metrics.elapsed_ns();
let elapsed_str = if elapsed >= 1_000_000_000 {
format!("{:.2}s", elapsed as f64 / 1_000_000_000.0)
} else if elapsed >= 1_000_000 {
format!("{:.2}ms", elapsed as f64 / 1_000_000.0)
} else if elapsed >= 1_000 {
format!("{:.2}µs", elapsed as f64 / 1_000.0)
} else {
format!("{}ns", elapsed)
};
let mut extra = String::new();
let scanned = metrics.edges_scanned();
if scanned > 0 {
extra.push_str(&format!(", scanned: {}", scanned));
}
let skipped = metrics.skipped_rows();
if skipped > 0 {
extra.push_str(&format!(", skipped: {}", skipped));
}
format!("rows: {}, batches: {}, elapsed: {}{}", rows, batches, elapsed_str, extra)
}
fn format_analyze_plan(
plan: &dyn ExecOperator,
output: &mut String,
prefix: &str,
redact_volatile_explain_attrs: bool,
) {
let name = plan.name();
let properties = plan.attrs();
let context = plan.required_context();
let _ = write!(output, "{} [ctx: {}]", name, context.short_name());
if !properties.is_empty() {
let _ = write!(output, " [");
for (i, (key, value)) in properties.iter().enumerate() {
if i > 0 {
let _ = write!(output, ", ");
}
let _ = write!(output, "{key}: {value}");
}
let _ = write!(output, "]");
}
if let Some(metrics) = plan.metrics() {
let _ =
write!(output, " {{{}}}", format_metrics_text(metrics, redact_volatile_explain_attrs));
}
let _ = writeln!(output);
let expressions = plan.expressions();
for (role, expr) in &expressions {
let embedded = expr.embedded_operators();
if !embedded.is_empty() {
for (embed_role, embed_plan) in &embedded {
let _ = write!(output, "{} {}.{}: ", prefix, role, embed_role);
format_analyze_plan(
embed_plan.as_ref(),
output,
&format!("{} ", prefix),
redact_volatile_explain_attrs,
);
}
}
}
let children = plan.children();
if !children.is_empty() {
let child_prefix = format!("{}{:width$}", prefix, "", width = INDENT_WIDTH);
for child in children.iter() {
let _ = write!(output, "{}", child_prefix);
format_analyze_plan(
child.as_ref(),
output,
&child_prefix,
redact_volatile_explain_attrs,
);
}
}
}
fn format_analyze_plan_json(
plan: &dyn ExecOperator,
redact_volatile_explain_attrs: bool,
) -> Object {
let mut obj = Object::default();
obj.insert("operator", Value::String(plan.name().into()));
obj.insert("context", Value::String(plan.required_context().short_name().into()));
let attrs = plan.attrs();
if !attrs.is_empty() {
let mut attrs_obj = Object::default();
for (key, value) in attrs {
attrs_obj.insert(key, Value::String(value.into()));
}
obj.insert("attributes", Value::Object(attrs_obj));
}
if let Some(metrics) = plan.metrics() {
let mut metrics_obj = Object::default();
metrics_obj.insert("output_rows", Value::from(metrics.output_rows() as i64));
if !redact_volatile_explain_attrs {
metrics_obj.insert("output_batches", Value::from(metrics.output_batches() as i64));
metrics_obj.insert("elapsed_ns", Value::from(metrics.elapsed_ns() as i64));
let scanned = metrics.edges_scanned();
if scanned > 0 {
metrics_obj.insert("edges_scanned", Value::from(scanned as i64));
}
let skipped = metrics.skipped_rows();
if skipped > 0 {
metrics_obj.insert("skipped_rows", Value::from(skipped as i64));
}
}
obj.insert("metrics", Value::Object(metrics_obj));
}
let expressions = plan.expressions();
if !expressions.is_empty() {
let exprs_arr: Vec<Value> = expressions
.iter()
.map(|(role, expr)| {
let mut expr_obj = Object::default();
expr_obj.insert("role", Value::String((*role).into()));
expr_obj.insert("sql", Value::String(expr.to_sql().into()));
let embedded = expr.embedded_operators();
if !embedded.is_empty() {
let embedded_arr: Vec<Value> = embedded
.iter()
.map(|(embed_role, embed_plan)| {
let mut e = Object::default();
e.insert("role", Value::String((*embed_role).into()));
e.insert(
"plan",
Value::Object(format_analyze_plan_json(
embed_plan.as_ref(),
redact_volatile_explain_attrs,
)),
);
Value::Object(e)
})
.collect();
expr_obj.insert("embedded_operators", Value::Array(Array::from(embedded_arr)));
}
Value::Object(expr_obj)
})
.collect();
obj.insert("expressions", Value::Array(Array::from(exprs_arr)));
}
let children = plan.children();
if !children.is_empty() {
let children_array: Vec<Value> = children
.iter()
.map(|child| {
Value::Object(format_analyze_plan_json(
child.as_ref(),
redact_volatile_explain_attrs,
))
})
.collect();
obj.insert("children", Value::Array(Array::from(children_array)));
}
obj
}