surrealdb-core 3.3.1

//! EXPLAIN and EXPLAIN ANALYZE operators.
//!
//! - [`ExplainPlan`] formats a query plan without executing it (read-only).
//! - [`AnalyzePlan`] executes the plan, drains it to completion, then formats the plan tree
//!   together with collected [`OperatorMetrics`].

use std::fmt::Write;
use std::sync::Arc;

use common::future::stream::{self, Yielder};
use futures::StreamExt;
use surrealdb_types::ToSql;

use crate::exec::context::{ContextLevel, ExecutionContext};
use crate::exec::{
	AccessMode, CardinalityHint, ExecOperator, FlowResult, OperatorMetrics, OutputShape,
	ValueBatch, ValueBatchStream, buffer_stream,
};
use crate::expr::{ControlFlow, ExplainFormat};
use crate::val::{Array, Object, Value};

/// Number of spaces used per indentation level in text plan output.
const INDENT_WIDTH: usize = 4;

/// EXPLAIN operator - formats an execution plan as text.
///
/// This operator wraps an inner statement's planned content and returns
/// the formatted execution plan as a string value.
#[derive(Debug)]
pub struct ExplainPlan {
	/// The inner statement's planned content
	pub plan: Arc<dyn ExecOperator>,
	/// The output format (currently only Text is supported)
	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 {
		// EXPLAIN doesn't need database context - it just formats the plan
		ContextLevel::Root
	}

	fn access_mode(&self) -> AccessMode {
		// EXPLAIN is always read-only - it doesn't execute the inner statement
		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(futures::stream::once(std::future::ready(Ok(ValueBatch::new(vec![output]))))))
	}

	fn output_shape(&self) -> OutputShape {
		// EXPLAIN returns a single scalar value (text or JSON)
		OutputShape::Scalar
	}
}

// =========================================================================
// EXPLAIN ANALYZE
// =========================================================================

/// EXPLAIN ANALYZE operator - executes the plan, collects metrics, then
/// formats the plan tree with runtime statistics.
///
/// Unlike [`ExplainPlan`], this operator actually executes the inner plan,
/// draining all batches to completion so that every operator's metrics are
/// populated. It then walks the operator tree exactly like `ExplainPlan`
/// but includes elapsed time, row counts, and batch counts.
#[derive(Debug)]
pub struct AnalyzePlan {
	/// The inner statement's planned content
	pub plan: Arc<dyn ExecOperator>,
	/// The output format
	pub format: ExplainFormat,
	/// When true, elapsed durations are omitted from the output, making
	/// it deterministic for test assertions.
	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 {
		// We actually execute the inner plan, so inherit its requirements
		self.plan.required_context()
	}

	fn access_mode(&self) -> AccessMode {
		// We execute the inner plan, so inherit its access mode
		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> {
		// Enable metrics on all operators before execution so that
		// monitor_stream wraps each stream with timing/counting.
		self.plan.enable_metrics();

		// Execute the inner plan to get its stream
		let mut inner_stream = buffer_stream(
			self.plan.execute(ctx)?,
			self.plan.access_mode(),
			self.plan.cardinality_hint(),
			ctx.root().ctx.config.exec.operator_buffer_size,
		);
		let plan = Arc::clone(&self.plan);
		let format = self.format;
		let redact_volatile_explain_attrs = self.redact_volatile_explain_attrs;

		// Create a stream that first drains the inner plan, then formats output
		let stream = stream::try_async_stream(async move |mut yielder: Yielder<_>| {
			// Drain all batches from the inner plan so metrics are populated
			let mut total_rows: u64 = 0;
			while let Some(batch_result) = inner_stream.next().await {
				match batch_result {
					Ok(batch) => {
						total_rows += batch.len() as u64;
					}
					// Flow control signals mean the inner plan stopped early.
					// Stop draining and format the metrics we've collected so far.
					Err(ControlFlow::Break | ControlFlow::Return(_)) => break,
					// Continue means skip this iteration, keep draining.
					Err(ControlFlow::Continue) => continue,
					// Only actual errors should propagate.
					Err(e @ ControlFlow::Err(_)) => Err(e)?,
				}
			}

			// Now format the plan with metrics
			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)
				}
			};

			yielder.emit(ValueBatch::new(vec![output])).await;
			Ok(())
		});

		Ok(Box::pin(stream))
	}

	fn output_shape(&self) -> OutputShape {
		OutputShape::Scalar
	}
}

// =========================================================================
// Text Formatting
// =========================================================================

/// Format an execution plan node as a text tree
fn format_execution_plan(plan: &dyn ExecOperator, output: &mut String, prefix: &str) {
	// Get operator name and properties
	let name = plan.name();
	let properties = plan.attrs();

	// Show context level
	let context = plan.required_context();
	let _ = write!(output, "{} [ctx: {}]", name, context.short_name());

	// Show the output partitioning only when the operator produces more than one
	// stream, so a plan built entirely from single-partition operators renders
	// exactly as it did before partitioning existed.
	let partitioning = plan.output_partitioning();
	if !partitioning.is_single() {
		let _ = write!(output, " [partitions: {}]", partitioning.count());
	}

	// Show properties if any
	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);

	// Format expressions that contain embedded operators
	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));
			}
		}
	}

	// Format children with indentation
	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);
		}
	}
}

// =========================================================================
// JSON Formatting
// =========================================================================

/// Format an execution plan node as a JSON object
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()));

	// Emitted only when the operator produces more than one stream, so a plan of
	// single-partition operators renders the same keys it always did.
	let partitioning = plan.output_partitioning();
	if !partitioning.is_single() {
		obj.insert("partitions", Value::from(partitioning.count() as i64));
	}

	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
}

// =========================================================================
// ANALYZE Formatters (include metrics)
// =========================================================================

/// Format metrics as a human-readable string fragment.
///
/// When `redact_volatile_explain_attrs` is true, elapsed time and batch counts are
/// omitted so the output is deterministic for test assertions.
fn format_metrics_text(metrics: &OperatorMetrics, redact_volatile_explain_attrs: bool) -> String {
	let rows = metrics.output_rows();

	// A bitmap branch that declined (budget overflow or a gate) is flagged so
	// its `rows: 0` is distinguishable from a genuinely empty candidate set.
	// Deterministic for a given dataset and configuration, so never redacted.
	let dropped = if metrics.branch_dropped() {
		", dropped: true"
	} else {
		""
	};

	if redact_volatile_explain_attrs {
		return format!("rows: {}{}", rows, dropped);
	}

	let batches = metrics.output_batches();
	let elapsed = metrics.elapsed_ns();

	// Format elapsed time in the most readable unit
	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)
	};

	// Scan-side counters that depend on how far the scan progressed (volatile
	// like batches/elapsed), rendered only when non-zero. `scanned` is the
	// denominator for graph-traversal filter selectivity (matched = rows);
	// `skipped` is the TopK threshold-pushdown reject count; `fetched` is the
	// count of candidate records evaluated inside an ANN search (#548 drives
	// it to zero for index-covered predicates).
	let mut extra = String::new();
	let scanned = metrics.edges_scanned();
	if scanned > 0 {
		extra.push_str(&format!(", scanned: {}", scanned));
	}
	// The storage-tier split of `scanned`: entries served from packed
	// adjacency blocks vs per-edge pointer keys.
	let block_hits = metrics.block_hits();
	if block_hits > 0 {
		extra.push_str(&format!(", block_hits: {}", block_hits));
	}
	let delta_hits = metrics.delta_hits();
	if delta_hits > 0 {
		extra.push_str(&format!(", delta_hits: {}", delta_hits));
	}
	let cache_hits = metrics.cache_hits();
	if cache_hits > 0 {
		extra.push_str(&format!(", cache_hits: {}", cache_hits));
	}
	let cache_misses = metrics.cache_misses();
	if cache_misses > 0 {
		extra.push_str(&format!(", cache_misses: {}", cache_misses));
	}
	let props_evals = metrics.props_evals();
	if props_evals > 0 {
		extra.push_str(&format!(", props_evals: {}", props_evals));
	}
	let props_fallbacks = metrics.props_fallbacks();
	if props_fallbacks > 0 {
		extra.push_str(&format!(", props_fallbacks: {}", props_fallbacks));
	}
	let skipped = metrics.skipped_rows();
	if skipped > 0 {
		extra.push_str(&format!(", skipped: {}", skipped));
	}
	let fetched = metrics.records_fetched();
	if fetched > 0 {
		extra.push_str(&format!(", fetched: {}", fetched));
	}
	format!("rows: {}, batches: {}, elapsed: {}{}{}", rows, batches, elapsed_str, extra, dropped)
}

/// Format an execution plan node as a text tree with metrics.
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();

	// Show context level
	let context = plan.required_context();
	let _ = write!(output, "{} [ctx: {}]", name, context.short_name());

	// Show the output partitioning only when the operator produces more than one
	// stream, so a plan built entirely from single-partition operators renders
	// exactly as it did before partitioning existed.
	let partitioning = plan.output_partitioning();
	if !partitioning.is_single() {
		let _ = write!(output, " [partitions: {}]", partitioning.count());
	}

	// Show properties if any
	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, "]");
	}

	// Show metrics if available
	if let Some(metrics) = plan.metrics() {
		let _ =
			write!(output, " {{{}}}", format_metrics_text(metrics, redact_volatile_explain_attrs));
	}

	let _ = writeln!(output);

	// Format expressions with embedded operators (with metrics)
	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,
				);
			}
		}
	}

	// Format children with indentation
	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,
			);
		}
	}
}

/// Format an execution plan node as a JSON object with metrics.
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()));

	// Emitted only when the operator produces more than one stream, so a plan of
	// single-partition operators renders the same keys it always did.
	let partitioning = plan.output_partitioning();
	if !partitioning.is_single() {
		obj.insert("partitions", Value::from(partitioning.count() as i64));
	}

	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));
		// Emitted only when a bitmap branch declined, so its `output_rows: 0`
		// is distinguishable from a genuinely empty candidate set.
		// Deterministic for a given dataset and configuration, so never
		// redacted.
		if metrics.branch_dropped() {
			metrics_obj.insert("dropped", Value::Bool(true));
		}
		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));
			// Volatile (progress/publish-timing dependent); rendered only when
			// non-zero. `edges_scanned` is the graph-traversal selectivity
			// denominator (matched = output_rows).
			let scanned = metrics.edges_scanned();
			if scanned > 0 {
				metrics_obj.insert("edges_scanned", Value::from(scanned as i64));
			}
			let block_hits = metrics.block_hits();
			if block_hits > 0 {
				metrics_obj.insert("block_hits", Value::from(block_hits as i64));
			}
			let delta_hits = metrics.delta_hits();
			if delta_hits > 0 {
				metrics_obj.insert("delta_hits", Value::from(delta_hits as i64));
			}
			let cache_hits = metrics.cache_hits();
			if cache_hits > 0 {
				metrics_obj.insert("cache_hits", Value::from(cache_hits as i64));
			}
			let cache_misses = metrics.cache_misses();
			if cache_misses > 0 {
				metrics_obj.insert("cache_misses", Value::from(cache_misses as i64));
			}
			let props_evals = metrics.props_evals();
			if props_evals > 0 {
				metrics_obj.insert("props_evals", Value::from(props_evals as i64));
			}
			let props_fallbacks = metrics.props_fallbacks();
			if props_fallbacks > 0 {
				metrics_obj.insert("props_fallbacks", Value::from(props_fallbacks as i64));
			}
			let skipped = metrics.skipped_rows();
			if skipped > 0 {
				metrics_obj.insert("skipped_rows", Value::from(skipped as i64));
			}
			// Deliberately rendered even at zero — unlike the counters above,
			// `records_fetched: 0` is itself the claim worth asserting: an ANN
			// search whose predicate was fully served by the allow-list bitmap
			// performed no in-traversal record fetches (#548).
			metrics_obj.insert("records_fetched", Value::from(metrics.records_fetched() 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)));
	}

	// Add children if any
	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
}