spiral/scandal/
profile.rs

1//! Bounded, collector-independent observations of the shared execution graph.
2
3use std::collections::BTreeMap;
4use std::fmt;
5
6use arrow_schema::Schema;
7use serde::{Deserialize, Serialize};
8
9use super::ScandalError;
10use super::plan::PlanDescription;
11
12/// Reserved request metadata, separate from the logical row schema.
13pub const EXECUTION_OPTIONS_KEY: &str = "spiral.execution.options";
14
15/// Metadata marker allowing clients to skip decoding disabled profiles.
16pub const PROFILE_METADATA_KEY: &str = "spiral.execution.profiling";
17
18pub use super::api::ExecutionOptions;
19
20#[expect(
21    clippy::derivable_impls,
22    reason = "Defaults are SDK behavior on a contract-generated request type"
23)]
24impl Default for ExecutionOptions {
25    fn default() -> Self {
26        Self {
27            profile: false,
28            caller: None,
29            partition: None,
30        }
31    }
32}
33
34impl ExecutionOptions {
35    /// Attach options without changing fields or application metadata.
36    pub fn with_schema(&self, schema: &Schema) -> Result<Schema, ScandalError> {
37        self.validate()?;
38        let mut metadata = schema.metadata().clone();
39        metadata.insert(
40            EXECUTION_OPTIONS_KEY.into(),
41            serde_json::to_string(self).map_err(invalid)?,
42        );
43        Ok(schema.clone().with_metadata(metadata))
44    }
45
46    /// Read options from a request; absent options disable profiling.
47    pub fn from_schema(schema: &Schema) -> Result<Self, ScandalError> {
48        let Some(value) = schema.metadata().get(EXECUTION_OPTIONS_KEY) else {
49            return Ok(Self::default());
50        };
51        if value.len() > 1024 {
52            return Err(invalid("execution options exceed 1 KiB"));
53        }
54        let options: Self = serde_json::from_str(value).map_err(invalid)?;
55        options.validate()?;
56        Ok(options)
57    }
58
59    pub(crate) fn validate(&self) -> Result<(), ScandalError> {
60        if self.caller.as_ref().is_some_and(|caller| {
61            caller.len() > 128
62                || !caller
63                    .bytes()
64                    .all(|c| c.is_ascii_alphanumeric() || b"-_.:".contains(&c))
65        }) {
66            return Err(invalid(
67                "execution caller must be an opaque identifier of at most 128 ASCII characters",
68            ));
69        }
70        Ok(())
71    }
72}
73
74/// Aggregate observations of one physical node across batches. Durations are nanoseconds.
75/// Elapsed host time includes awaits; it is not CPU time or device event time.
76#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
77#[serde(rename_all = "camelCase")]
78pub struct NodeProfile {
79    /// Time waiting for a declared backend limit; absent when unmeasured.
80    pub admission_wait_ns: Option<u64>,
81    /// Whether this executor directly observed invocations of the node.
82    pub measured: bool,
83    /// Number of completed or failed invocations observed.
84    pub calls: u64,
85    /// Rows presented to this node.
86    pub input_rows: u64,
87    /// Rows returned by successful invocations.
88    pub output_rows: u64,
89    /// Referenced Arrow or Vortex buffer bytes; these are not unique allocations.
90    pub input_bytes: u64,
91    /// Returned buffer bytes, which can alias input buffers.
92    pub output_bytes: u64,
93    /// Inclusive host elapsed time, summed across invocations, including asynchronous waits.
94    pub elapsed_ns: u64,
95    /// Time a completed batch waits for ordered delivery.
96    pub ordered_wait_ns: u64,
97    /// BlobIo read/stat calls, before transport coalescing/cache effects.
98    pub io_calls: Option<u64>,
99    /// Bytes returned through BlobIo; not necessarily bytes fetched from object storage.
100    pub io_bytes: Option<u64>,
101    /// Inclusive BlobIo elapsed time; a subset of this node's elapsed time.
102    pub io_elapsed_ns: Option<u64>,
103    /// Confirmed copied bytes, absent when the backend cannot measure them.
104    pub copied_bytes: Option<u64>,
105    /// Device event time, absent when the implementation exposes no timing events.
106    pub device_ns: Option<u64>,
107    /// Synchronization time, absent when it cannot be separated from host elapsed time.
108    pub synchronization_ns: Option<u64>,
109    /// Backend session creations, absent without backend instrumentation.
110    pub sessions_created: Option<u64>,
111    /// Backend session reuses, absent without backend instrumentation.
112    pub sessions_reused: Option<u64>,
113}
114
115/// A bounded aggregate profile for one execution, embedded in its plan description.
116#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
117#[serde(rename_all = "camelCase")]
118pub struct ExecutionProfile {
119    /// `running`, `completed`, `failed`, or `cancelled`.
120    pub status: String,
121    /// Existing operation carrying the report, such as `evaluate` or `scan`.
122    pub operation: String,
123    /// Caller correlation supplied in the execution request.
124    pub caller: Option<String>,
125    /// Caller partition, if supplied.
126    pub partition: Option<u32>,
127    /// Physical-node IDs from the enclosing plan. No per-row event history is retained.
128    pub nodes: BTreeMap<String, NodeProfile>,
129    /// Nodes omitted by the 1024-node aggregate limit.
130    pub omitted_nodes: u64,
131    /// Cumulative profile observations omitted by this delivery merge, if any.
132    pub delivery_omissions: u64,
133    /// Elapsed time waiting for the next input batch, inclusive of upstream I/O/work.
134    pub input_wait_ns: u64,
135    /// Time between yielding an output and the adapter polling again; not remote user time.
136    pub consumer_wait_ns: u64,
137    /// Largest number of active or buffered batches held by this projection.
138    pub peak_in_flight_batches: Option<u64>,
139    /// Peak sum of admitted input batch buffer estimates. May double-count shared buffers.
140    pub peak_input_buffer_bytes: Option<u64>,
141    /// Peak delivery reservation reported by this operation; not RSS or total host memory.
142    pub peak_delivery_reserved_bytes: Option<u64>,
143    /// Device allocation reservations, unknown until a backend supplies accounting.
144    pub peak_device_reserved_bytes: Option<u64>,
145    /// Time waiting for declared backend limits; generic memory admission is not measured.
146    pub admission_wait_ns: Option<u64>,
147    /// Physical node whose invocation failed, without the error's payload or URL.
148    pub failed_node: Option<String>,
149    /// Sanitized contract error classification; messages and payloads are excluded.
150    pub error_code: Option<String>,
151    /// Whether the consumer stopped reading before a terminal host report was received.
152    pub consumer_cancelled: bool,
153}
154
155impl fmt::Display for ExecutionProfile {
156    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
157        write!(
158            f,
159            "analyze status={} operation={} input_wait_ns={} consumer_wait_ns={} peak_in_flight={} peak_input_buffer_bytes={}",
160            self.status,
161            self.operation,
162            self.input_wait_ns,
163            self.consumer_wait_ns,
164            self.peak_in_flight_batches
165                .map_or_else(|| "unknown".into(), |n| n.to_string()),
166            self.peak_input_buffer_bytes
167                .map_or_else(|| "unknown".into(), |n| n.to_string())
168        )?;
169        if let Some(caller) = &self.caller {
170            write!(f, " caller={caller}")?;
171        }
172        if let Some(partition) = self.partition {
173            write!(f, " partition={partition}")?;
174        }
175        if self.delivery_omissions > 0 {
176            write!(f, " delivery_omissions={}", self.delivery_omissions)?;
177        }
178        for (id, node) in &self.nodes {
179            if !node.measured {
180                write!(f, "\n  {id} measurements=unknown")?;
181                continue;
182            }
183            write!(
184                f,
185                "\n  {id} calls={} rows={}->{} bytes={}->{} host_elapsed_ns={} ordered_wait_ns={} device_ns={}",
186                node.calls,
187                node.input_rows,
188                node.output_rows,
189                node.input_bytes,
190                node.output_bytes,
191                node.elapsed_ns,
192                node.ordered_wait_ns,
193                node.device_ns
194                    .map_or_else(|| "unknown".into(), |n| n.to_string())
195            )?;
196            if let Some(wait) = node.admission_wait_ns {
197                write!(f, " admission_wait_ns={wait}")?;
198            }
199            if let Some(calls) = node.io_calls {
200                write!(
201                    f,
202                    " io_calls={calls} io_bytes={} io_elapsed_ns={}",
203                    node.io_bytes.unwrap_or(0),
204                    node.io_elapsed_ns.unwrap_or(0)
205                )?;
206            }
207        }
208        if let Some(node) = &self.failed_node {
209            write!(f, "\n  failed_node={node}")?;
210        }
211        Ok(())
212    }
213}
214
215fn invalid(error: impl fmt::Display) -> ScandalError {
216    ScandalError::InvalidArgument {
217        message: error.to_string(),
218    }
219}
220
221/// Reconstruct a redacted execution error after a terminal failed profile frame.
222pub fn execution_error(plan: &PlanDescription) -> Option<ScandalError> {
223    let profile = plan.profile.as_ref().filter(|p| p.status == "failed")?;
224    let node = profile.failed_node.as_deref().unwrap_or("input");
225    let outputs = plan
226        .logical
227        .iter()
228        .filter_map(|n| n.properties.get("output"))
229        .cloned()
230        .collect::<Vec<_>>()
231        .join(",");
232    let message = format!(
233        "SPQL execution={} plan={} node={node} outputs=[{outputs}] failed",
234        plan.execution_id.as_deref().unwrap_or("unknown"),
235        plan.physical_id
236    );
237    Some(match profile.error_code.as_deref() {
238        Some("budget_exceeded") => ScandalError::BudgetExceeded { message },
239        Some("invalid_argument") => ScandalError::InvalidArgument { message },
240        Some("conflict") => ScandalError::Conflict { message },
241        Some("storage_unavailable") => ScandalError::StorageUnavailable { message },
242        Some("unavailable") => ScandalError::Unavailable { message },
243        Some("unauthenticated") => ScandalError::Unauthenticated { message },
244        Some("permission_denied") => ScandalError::PermissionDenied {
245            operation: "execute".into(),
246            resource: message,
247        },
248        Some("not_found") => ScandalError::NotFound { resource: message },
249        _ => ScandalError::Internal { message },
250    })
251}
spiral/scandal/profile.rs

spiral/scandal/
profile.rs
