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