query_planning/
error.rs

1//! Error taxonomy for the native query engine.
2//!
3//! Native errors stop at the `query-engine` API boundary as [`QueryError`].
4//! The DataFusion adapter wraps them in its own `algeon_datafusion::Error` so it
5//! can preserve native classification facts while exposing adapter-level
6//! planning and execution categories to DataFusion.
7
8use std::{
9    borrow::Cow,
10    collections::{BTreeMap, BTreeSet},
11    error::Error as StdError,
12    fmt,
13    sync::Arc,
14};
15
16use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error as _};
17use serde_json::{Value, json};
18
19pub use self::context::{AllocationSource, FailureContext, MemoryFailure, MemoryUsage, ResourceBoundary};
20use crate::{
21    correlation::{QueryId, TraceId},
22    triage::TriageCode,
23};
24
25mod context;
26
27// Shared so an error can be cloned: one failure may be reported to every
28// party waiting on the operation that failed.
29type SharedSourceError = Arc<dyn StdError + Send + Sync + 'static>;
30
31/// Native query-engine result type.
32pub type Result<T> = std::result::Result<T, QueryError>;
33
34/// Stable, machine-readable native-engine error codes.
35///
36/// Codes describe the closest actionable cause. [`QueryErrorKind`] and
37/// [`QueryErrorStatus`] are derived from the code; consumers should not infer
38/// those policies from variant names.
39#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
40pub enum QueryErrorCode {
41    /// Query execution was cancelled by the caller.
42    Cancelled,
43    /// A compiled pipeline contract refuses a plan that must execute natively.
44    ///
45    /// Identifies a capability rejection. Malformed plans report as `InvalidPlan`.
46    UnsupportedPipelineContract,
47    /// A specific operator within an otherwise recognizable plan is unsupported.
48    UnsupportedOperator,
49    /// An expression or expression form is unsupported.
50    UnsupportedExpression,
51    /// A data type, physical representation, or type combination is unsupported.
52    UnsupportedType,
53    /// The native IR violates executor invariants or cannot be validated.
54    ///
55    /// Use this when the plan is malformed for the native engine. A well-formed
56    /// plan outside the supported catalogue is reported with an `Unsupported*` code.
57    InvalidPlan,
58    /// Caller-supplied execution-attempt correlation is invalid or oversized.
59    InvalidAttemptCorrelation,
60    /// A service, backend, or device configuration value is invalid or
61    /// inconsistent with the discovered hardware.
62    ///
63    /// Unlike [`Self::InvalidPlan`], the input is deployment configuration, not
64    /// a query plan. The same configuration fails again until it changes.
65    InvalidConfiguration,
66    /// Runtime graph inputs violate an engine-owned graph-operation contract.
67    ///
68    /// This code describes materialized edge or side-input data. `InvalidPlan`
69    /// describes malformed native IR.
70    GraphInputInvalid,
71    /// cuDF rejected an input, including permanent `NotFound` conditions.
72    CudfInvalidInput,
73    /// cuDF reported a permanent unsupported capability.
74    CudfUnsupported,
75    /// A host materialization size computation overflowed the platform range.
76    /// The same request cannot be retried safely without changing its shape.
77    SizeOverflow,
78    /// One unsplittable Arrow host row exceeds the configured deterministic
79    /// materialization ceiling.
80    HostMaterializationLimitExceeded,
81    /// cuDF reported a temporary dependency/runtime failure.
82    CudfTemporaryFailure,
83    /// cuDF reported a persistent dependency/runtime failure.
84    ///
85    /// The failure has been observed to recur or is known to continue. Stop
86    /// automatic retries and escalate or wait for intervention.
87    CudfPersistentFailure,
88    /// cuDF reported a permanent dependency failure.
89    CudfDependencyFailure,
90    /// An iterative native algorithm reached its configured iteration bound.
91    AlgorithmDidNotConverge,
92    /// Source object storage denied access to the requested object or metadata.
93    SourceAccessDenied,
94    /// Source object storage lacked valid authentication credentials.
95    SourceUnauthenticated,
96    /// Source object storage could not find the requested object.
97    SourceObjectNotFound,
98    /// Source object storage failed with a retryable transport or service error.
99    SourceTemporaryFailure,
100    /// cuDF reported an invariant or execution-context violation.
101    CudfInvariantViolation,
102    /// The native admission controller could not reserve requested memory.
103    ///
104    /// Represents a temporary resource condition driving retry or backoff
105    /// decisions, separate from cuDF allocator failures.
106    MemoryReservationFailed,
107    /// The service admission layer rejected the query because shared bounded
108    /// capacity is saturated.
109    ServiceOverloaded,
110    /// The requested query cap cannot fit on any eligible device after
111    /// subtracting resident ownership.
112    QueryBudgetUnavailable,
113    /// The query domain rejected an actual or exact-required allocation over
114    /// the query's fixed memory cap.
115    QueryMemoryLimitExceeded,
116    /// A device allocation failed with confirmed physical-memory provenance.
117    DeviceOutOfMemory,
118    /// A device allocation failed without confirmed physical-memory provenance.
119    DeviceAllocationFailed,
120    /// A persistent device or CUDA runtime fault made the affected device
121    /// unavailable.
122    DeviceFault,
123    /// Cleanup or synchronization could not establish safe release of the
124    /// attempt's device resources.
125    CleanupFailed,
126    /// An owner, provenance, ledger, or stream-routing invariant was violated.
127    InternalInvariantViolation,
128    /// An internal native-engine bug or unreachable condition.
129    Internal,
130}
131
132impl QueryErrorCode {
133    /// Every native error code supported by this engine release.
134    ///
135    /// This list keeps adapter projections and status policy mechanically
136    /// auditable when a new code is introduced.
137    pub const ALL: &'static [Self] = &[
138        Self::Cancelled,
139        Self::UnsupportedPipelineContract,
140        Self::UnsupportedOperator,
141        Self::UnsupportedExpression,
142        Self::UnsupportedType,
143        Self::InvalidPlan,
144        Self::InvalidAttemptCorrelation,
145        Self::InvalidConfiguration,
146        Self::GraphInputInvalid,
147        Self::CudfInvalidInput,
148        Self::CudfUnsupported,
149        Self::SizeOverflow,
150        Self::HostMaterializationLimitExceeded,
151        Self::CudfTemporaryFailure,
152        Self::CudfPersistentFailure,
153        Self::CudfDependencyFailure,
154        Self::AlgorithmDidNotConverge,
155        Self::SourceAccessDenied,
156        Self::SourceUnauthenticated,
157        Self::SourceObjectNotFound,
158        Self::SourceTemporaryFailure,
159        Self::CudfInvariantViolation,
160        Self::MemoryReservationFailed,
161        Self::ServiceOverloaded,
162        Self::QueryBudgetUnavailable,
163        Self::QueryMemoryLimitExceeded,
164        Self::DeviceOutOfMemory,
165        Self::DeviceAllocationFailed,
166        Self::DeviceFault,
167        Self::CleanupFailed,
168        Self::InternalInvariantViolation,
169        Self::Internal,
170    ];
171
172    /// Stable snake-case code label used in reports and error text.
173    pub const fn as_str(self) -> &'static str {
174        match self {
175            Self::Cancelled => "cancelled",
176            Self::UnsupportedPipelineContract => "unsupported_pipeline_contract",
177            Self::UnsupportedOperator => "unsupported_operator",
178            Self::UnsupportedExpression => "unsupported_expression",
179            Self::UnsupportedType => "unsupported_type",
180            Self::InvalidPlan => "invalid_plan",
181            Self::InvalidAttemptCorrelation => "invalid_attempt_correlation",
182            Self::InvalidConfiguration => "invalid_configuration",
183            Self::GraphInputInvalid => "graph_input_invalid",
184            Self::CudfInvalidInput => "cudf_invalid_input",
185            Self::CudfUnsupported => "cudf_unsupported",
186            Self::SizeOverflow => "size_overflow",
187            Self::HostMaterializationLimitExceeded => "host_materialization_limit_exceeded",
188            Self::CudfTemporaryFailure => "cudf_temporary_failure",
189            Self::CudfPersistentFailure => "cudf_persistent_failure",
190            Self::CudfDependencyFailure => "cudf_dependency_failure",
191            Self::AlgorithmDidNotConverge => "algorithm_did_not_converge",
192            Self::SourceAccessDenied => "source_access_denied",
193            Self::SourceUnauthenticated => "source_unauthenticated",
194            Self::SourceObjectNotFound => "source_object_not_found",
195            Self::SourceTemporaryFailure => "source_temporary_failure",
196            Self::CudfInvariantViolation => "cudf_invariant_violation",
197            Self::MemoryReservationFailed => "memory_reservation_failed",
198            Self::ServiceOverloaded => "service_overloaded",
199            Self::QueryBudgetUnavailable => "query_budget_unavailable",
200            Self::QueryMemoryLimitExceeded => "query_memory_limit_exceeded",
201            Self::DeviceOutOfMemory => "device_out_of_memory",
202            Self::DeviceAllocationFailed => "device_allocation_failed",
203            Self::DeviceFault => "device_fault",
204            Self::CleanupFailed => "cleanup_failed",
205            Self::InternalInvariantViolation => "internal_invariant_violation",
206            Self::Internal => "internal",
207        }
208    }
209
210    /// Returns the stable coarse error category for this code.
211    pub const fn kind(self) -> QueryErrorKind {
212        kind_for_code(self)
213    }
214
215    /// Returns the retry/remediation status for this code.
216    pub const fn status(self) -> QueryErrorStatus {
217        status_for_code(self)
218    }
219}
220
221impl fmt::Display for QueryErrorCode {
222    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
223        f.write_str(self.as_str())
224    }
225}
226
227/// Coarse native-engine error category.
228#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
229#[non_exhaustive]
230pub enum QueryErrorKind {
231    /// Caller-requested cancellation stopped execution.
232    Cancelled,
233    /// Planning, validation, or IR construction failure.
234    Planning,
235    /// Caller-provided input is invalid for the requested operation.
236    InvalidInput,
237    /// Requested plan, operator, expression, or type is not supported.
238    Unsupported,
239    /// Memory or other bounded resource was exhausted or could not be reserved.
240    ResourceExhausted,
241    /// cuDF or another dependency failed outside resource-exhaustion handling.
242    DependencyFailure,
243    /// A native-engine or cuDF invariant was violated.
244    InvariantViolation,
245    /// Internal bug or unreachable condition.
246    Internal,
247}
248
249impl QueryErrorKind {
250    /// Stable snake-case category label used in reports and error text.
251    pub(crate) const fn as_str(self) -> &'static str {
252        match self {
253            Self::Cancelled => "cancelled",
254            Self::Planning => "planning",
255            Self::InvalidInput => "invalid_input",
256            Self::Unsupported => "unsupported",
257            Self::ResourceExhausted => "resource_exhausted",
258            Self::DependencyFailure => "dependency_failure",
259            Self::InvariantViolation => "invariant_violation",
260            Self::Internal => "internal",
261        }
262    }
263}
264
265impl fmt::Display for QueryErrorKind {
266    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
267        f.write_str(self.as_str())
268    }
269}
270
271/// Retry and remediation status for native-engine errors.
272///
273/// Provides coarse source classification. Attempt retry advice is derived
274/// only after terminal evidence is sealed.
275#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
276pub enum QueryErrorStatus {
277    /// The same request will not succeed without changing its query, input,
278    /// configuration, or code. Do not retry it automatically.
279    Permanent,
280    /// The failure may clear without external repair. A sealed attempt report
281    /// still needs terminal evidence before it can advise backoff or retry.
282    Temporary,
283    /// A dependency/runtime failure has been observed to recur or is known to
284    /// continue. Stop automatic retries and escalate or wait for intervention.
285    ///
286    /// This remains distinct from [`Self::Permanent`]. It records a continuing
287    /// dependency failure, while `Permanent` records a request that must change.
288    /// Neither status alone is retry advice.
289    Persistent,
290}
291
292impl QueryErrorStatus {
293    /// Stable snake-case status label used in reports and error text.
294    pub const fn as_str(self) -> &'static str {
295        match self {
296            Self::Permanent => "permanent",
297            Self::Temporary => "temporary",
298            Self::Persistent => "persistent",
299        }
300    }
301}
302
303impl fmt::Display for QueryErrorStatus {
304    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
305        f.write_str(self.as_str())
306    }
307}
308
309/// Maps each native error code to its stable coarse category.
310///
311/// `kind` is the broad bucket used for diagnostics and aggregation. It is less
312/// specific than [`QueryErrorCode`] and should remain stable for consumers that
313/// group failures without depending on every code variant.
314pub(crate) const fn kind_for_code(code: QueryErrorCode) -> QueryErrorKind {
315    match code {
316        QueryErrorCode::Cancelled => QueryErrorKind::Cancelled,
317        QueryErrorCode::UnsupportedPipelineContract
318        | QueryErrorCode::UnsupportedOperator
319        | QueryErrorCode::UnsupportedExpression
320        | QueryErrorCode::UnsupportedType => QueryErrorKind::Unsupported,
321        QueryErrorCode::InvalidPlan => QueryErrorKind::Planning,
322        QueryErrorCode::InvalidAttemptCorrelation => QueryErrorKind::InvalidInput,
323        QueryErrorCode::InvalidConfiguration => QueryErrorKind::InvalidInput,
324        QueryErrorCode::GraphInputInvalid => QueryErrorKind::InvalidInput,
325        QueryErrorCode::CudfInvalidInput => QueryErrorKind::InvalidInput,
326        QueryErrorCode::CudfUnsupported => QueryErrorKind::Unsupported,
327        QueryErrorCode::SizeOverflow
328        | QueryErrorCode::HostMaterializationLimitExceeded
329        | QueryErrorCode::MemoryReservationFailed
330        | QueryErrorCode::ServiceOverloaded
331        | QueryErrorCode::QueryBudgetUnavailable
332        | QueryErrorCode::QueryMemoryLimitExceeded
333        | QueryErrorCode::DeviceOutOfMemory
334        | QueryErrorCode::DeviceAllocationFailed => QueryErrorKind::ResourceExhausted,
335        QueryErrorCode::CudfTemporaryFailure
336        | QueryErrorCode::CudfPersistentFailure
337        | QueryErrorCode::CudfDependencyFailure
338        | QueryErrorCode::AlgorithmDidNotConverge
339        | QueryErrorCode::SourceAccessDenied
340        | QueryErrorCode::SourceUnauthenticated
341        | QueryErrorCode::SourceObjectNotFound
342        | QueryErrorCode::SourceTemporaryFailure
343        | QueryErrorCode::DeviceFault
344        | QueryErrorCode::CleanupFailed => QueryErrorKind::DependencyFailure,
345        QueryErrorCode::CudfInvariantViolation | QueryErrorCode::InternalInvariantViolation => {
346            QueryErrorKind::InvariantViolation
347        },
348        QueryErrorCode::Internal => QueryErrorKind::Internal,
349    }
350}
351
352/// Maps each native error code to its stable retry/remediation status.
353///
354/// Attempt retry advice is derived only after terminal evidence is sealed.
355/// [`QueryErrorStatus::Persistent`] means a repeated or known-continuing
356/// dependency failure that requires escalation or intervention, while
357/// [`QueryErrorStatus::Permanent`] means the request, input, configuration, or
358/// code must change before the same operation can succeed.
359pub(crate) const fn status_for_code(code: QueryErrorCode) -> QueryErrorStatus {
360    match code {
361        QueryErrorCode::Cancelled => QueryErrorStatus::Permanent,
362        QueryErrorCode::CudfTemporaryFailure
363        | QueryErrorCode::SourceTemporaryFailure
364        | QueryErrorCode::MemoryReservationFailed
365        | QueryErrorCode::ServiceOverloaded
366        | QueryErrorCode::QueryBudgetUnavailable
367        | QueryErrorCode::DeviceOutOfMemory
368        | QueryErrorCode::DeviceAllocationFailed => QueryErrorStatus::Temporary,
369        QueryErrorCode::CudfPersistentFailure => QueryErrorStatus::Persistent,
370        QueryErrorCode::UnsupportedPipelineContract
371        | QueryErrorCode::UnsupportedOperator
372        | QueryErrorCode::UnsupportedExpression
373        | QueryErrorCode::UnsupportedType
374        | QueryErrorCode::InvalidPlan
375        | QueryErrorCode::InvalidAttemptCorrelation
376        | QueryErrorCode::InvalidConfiguration
377        | QueryErrorCode::GraphInputInvalid
378        | QueryErrorCode::CudfInvalidInput
379        | QueryErrorCode::CudfUnsupported
380        | QueryErrorCode::SizeOverflow
381        | QueryErrorCode::HostMaterializationLimitExceeded
382        | QueryErrorCode::CudfDependencyFailure
383        | QueryErrorCode::AlgorithmDidNotConverge
384        | QueryErrorCode::SourceAccessDenied
385        | QueryErrorCode::SourceUnauthenticated
386        | QueryErrorCode::SourceObjectNotFound
387        | QueryErrorCode::CudfInvariantViolation
388        | QueryErrorCode::QueryMemoryLimitExceeded
389        | QueryErrorCode::InternalInvariantViolation => QueryErrorStatus::Permanent,
390        QueryErrorCode::DeviceFault | QueryErrorCode::CleanupFailed => QueryErrorStatus::Persistent,
391        QueryErrorCode::Internal => QueryErrorStatus::Permanent,
392    }
393}
394
395pub(crate) const fn triage_code_for_query_error(code: QueryErrorCode) -> TriageCode {
396    match code {
397        QueryErrorCode::Cancelled => TriageCode::Cancelled,
398        QueryErrorCode::UnsupportedPipelineContract => TriageCode::UnsupportedPlanShape,
399        QueryErrorCode::UnsupportedOperator => TriageCode::UnsupportedOperator,
400        QueryErrorCode::UnsupportedExpression => TriageCode::UnsupportedExpression,
401        QueryErrorCode::UnsupportedType => TriageCode::UnsupportedType,
402        QueryErrorCode::InvalidPlan => TriageCode::InvalidPlan,
403        QueryErrorCode::InvalidAttemptCorrelation
404        | QueryErrorCode::InvalidConfiguration
405        | QueryErrorCode::GraphInputInvalid
406        | QueryErrorCode::CudfInvalidInput => TriageCode::InvalidInput,
407        QueryErrorCode::CudfUnsupported => TriageCode::UnsupportedOperator,
408        QueryErrorCode::SizeOverflow
409        | QueryErrorCode::HostMaterializationLimitExceeded
410        | QueryErrorCode::QueryMemoryLimitExceeded
411        | QueryErrorCode::DeviceOutOfMemory
412        | QueryErrorCode::DeviceAllocationFailed => TriageCode::ResourceExhausted,
413        QueryErrorCode::CudfTemporaryFailure
414        | QueryErrorCode::CudfPersistentFailure
415        | QueryErrorCode::CudfDependencyFailure
416        | QueryErrorCode::AlgorithmDidNotConverge => TriageCode::DependencyFailure,
417        QueryErrorCode::CudfInvariantViolation => TriageCode::InvariantViolation,
418        QueryErrorCode::ServiceOverloaded => TriageCode::ServiceOverloaded,
419        QueryErrorCode::Internal => TriageCode::Internal,
420        QueryErrorCode::SourceAccessDenied
421        | QueryErrorCode::SourceUnauthenticated
422        | QueryErrorCode::SourceObjectNotFound
423        | QueryErrorCode::SourceTemporaryFailure => TriageCode::SourceFailure,
424        QueryErrorCode::MemoryReservationFailed => TriageCode::MemoryReservationDenied,
425        QueryErrorCode::QueryBudgetUnavailable => TriageCode::MemoryReservationDenied,
426        QueryErrorCode::DeviceFault | QueryErrorCode::CleanupFailed => TriageCode::DependencyFailure,
427        QueryErrorCode::InternalInvariantViolation => TriageCode::InvariantViolation,
428    }
429}
430
431/// Failure code of a sealed attempt: a native error's own code, or one of the
432/// attempt lifecycle outcomes no `QueryError` carries.
433///
434/// The wire label is the native code's label and `Debug` prints the native
435/// code's name, so wrapping changes no report value.
436#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
437pub enum AttemptFailureCode {
438    /// Failure classified by the canonical native query error taxonomy.
439    Query(QueryErrorCode),
440    /// The backend was closing when the attempt needed service.
441    BackendClosing,
442    /// A DataFusion-owned failure terminated the attempt.
443    DatafusionError,
444    /// The result consumer dropped the attempt before normal completion.
445    Dropped,
446    /// Attempt execution unwound after a panic.
447    Panicked,
448    /// Stream completion could not establish safe resource release.
449    StreamSynchronizationFailed,
450    /// A cleanup callback was abandoned before it could report completion.
451    CleanupCallbackAbandoned,
452}
453
454impl AttemptFailureCode {
455    const LIFECYCLE: [Self; 6] = [
456        Self::BackendClosing,
457        Self::DatafusionError,
458        Self::Dropped,
459        Self::Panicked,
460        Self::StreamSynchronizationFailed,
461        Self::CleanupCallbackAbandoned,
462    ];
463
464    /// Wraps a native error code as an attempt failure without changing its classification.
465    pub const fn from_query_error_code(code: QueryErrorCode) -> Self {
466        Self::Query(code)
467    }
468
469    /// Stable label used by attempt reports for this failure code.
470    pub const fn as_str(self) -> &'static str {
471        match self {
472            Self::Query(code) => code.as_str(),
473            Self::BackendClosing => "backend_closing",
474            Self::DatafusionError => "datafusion_error",
475            Self::Dropped => "dropped",
476            Self::Panicked => "panicked",
477            Self::StreamSynchronizationFailed => "stream_synchronization_failed",
478            Self::CleanupCallbackAbandoned => "cleanup_callback_abandoned",
479        }
480    }
481
482    fn from_label(label: &str) -> Option<Self> {
483        QueryErrorCode::ALL
484            .iter()
485            .map(|code| Self::Query(*code))
486            .chain(Self::LIFECYCLE)
487            .find(|code| code.as_str() == label)
488    }
489}
490
491impl fmt::Debug for AttemptFailureCode {
492    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
493        match self {
494            Self::Query(code) => fmt::Debug::fmt(code, f),
495            Self::BackendClosing => f.write_str("BackendClosing"),
496            Self::DatafusionError => f.write_str("DatafusionError"),
497            Self::Dropped => f.write_str("Dropped"),
498            Self::Panicked => f.write_str("Panicked"),
499            Self::StreamSynchronizationFailed => f.write_str("StreamSynchronizationFailed"),
500            Self::CleanupCallbackAbandoned => f.write_str("CleanupCallbackAbandoned"),
501        }
502    }
503}
504
505impl Serialize for AttemptFailureCode {
506    fn serialize<S: Serializer>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error> {
507        serializer.serialize_str(self.as_str())
508    }
509}
510
511impl<'de> Deserialize<'de> for AttemptFailureCode {
512    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> {
513        let label = <Cow<'de, str>>::deserialize(deserializer)?;
514        Self::from_label(&label).ok_or_else(|| D::Error::custom(format!("unknown attempt failure code `{label}`")))
515    }
516}
517
518/// Machine-readable facts attached to a [`QueryError`].
519///
520/// Fact keys are static strings so they can be treated as a documented contract
521/// by tests, reports, and callers. Values are stringified with [`ToString`].
522/// Iteration is sorted by key because facts are stored in a [`BTreeMap`].
523///
524/// Decoration never panics: empty keys are ignored and duplicate inserts keep
525/// the first value. Rejected keys are reported deterministically under
526/// `fact_collision_keys` so a colliding decoration stays diagnosable without
527/// replacing the primary error's facts. Callers that own a key and must
528/// overwrite it use [`QueryError::with_query_id`], [`QueryError::with_trace_id`],
529/// or another explicit overwrite path, never a second [`QueryError::with_fact`].
530/// The native engine reserves `cudf_code`, `cudf_kind`, `cudf_status`,
531/// `cudf_operation`, and `cudf_message` for wrapped `cudf::Error` facts.
532#[derive(Debug, Clone, Default, PartialEq, Eq)]
533pub struct Facts {
534    values: BTreeMap<&'static str, String>,
535    collision_keys: BTreeSet<&'static str>,
536}
537
538/// Fact key recording rejected duplicate or empty decoration keys.
539pub(crate) const FACT_COLLISION_KEYS: &str = "fact_collision_keys";
540
541/// Placeholder recorded when a decoration key is empty.
542pub(crate) const EMPTY_FACT_KEY: &str = "<empty>";
543
544impl Facts {
545    pub(crate) fn new() -> Self {
546        Self::default()
547    }
548
549    /// Looks up a stored fact by name, returning `None` when absent.
550    pub fn get(&self, key: &str) -> Option<&str> {
551        self.values.get(key).map(String::as_str)
552    }
553
554    /// Iterates facts in sorted key order.
555    pub fn iter(&self) -> impl Iterator<Item = (&'static str, &str)> {
556        self.values.iter().map(|(key, value)| (*key, value.as_str()))
557    }
558
559    fn insert(&mut self, key: &'static str, value: String) {
560        if key.is_empty() {
561            self.record_collision(EMPTY_FACT_KEY);
562            return;
563        }
564        if key == FACT_COLLISION_KEYS || self.values.contains_key(key) {
565            self.record_collision(key);
566            return;
567        }
568        self.values.insert(key, value);
569    }
570
571    fn record_collision(&mut self, key: &'static str) {
572        self.collision_keys.insert(key);
573        let value = self.collision_keys.iter().copied().collect::<Vec<_>>().join(",");
574        self.values.insert(FACT_COLLISION_KEYS, value);
575    }
576
577    fn set(&mut self, key: &'static str, value: String) {
578        assert!(!key.is_empty(), "query-engine error fact key must not be empty");
579        self.values.insert(key, value);
580    }
581}
582
583/// Structured native-engine error with stable classification and facts.
584#[derive(Clone)]
585pub struct QueryError(Box<QueryErrorRepr>);
586
587#[derive(Clone)]
588struct QueryErrorRepr {
589    query_id: Option<QueryId>,
590    trace_id: Option<TraceId>,
591    code: QueryErrorCode,
592    operation: &'static str,
593    message: String,
594    facts: Box<Facts>,
595    context: FailureContext,
596    source: Option<SharedSourceError>,
597}
598
599impl QueryError {
600    /// Creates a native-engine error with a stable operation id and message.
601    ///
602    /// `operation` should be a stable dotted identifier such as
603    /// `query_engine.plan.execute`. `message` is human-facing diagnostic
604    /// text. Add machine-readable context with [`Self::with_fact`].
605    ///
606    /// # Panics
607    ///
608    /// Panics when `operation` or `message` is empty. Construction-time
609    /// identity must be present; later decoration ([`Self::with_fact`],
610    /// [`Self::with_source`]) never panics and keeps the first value.
611    pub fn new(code: QueryErrorCode, operation: &'static str, message: impl Into<String>) -> Self {
612        let message = message.into();
613        assert!(!operation.is_empty(), "query-engine error operation must not be empty");
614        assert!(!message.is_empty(), "query-engine error message must not be empty");
615        Self(Box::new(QueryErrorRepr {
616            query_id: None,
617            trace_id: None,
618            code,
619            operation,
620            message,
621            facts: Box::new(Facts::new()),
622            context: FailureContext::default(),
623            source: None,
624        }))
625    }
626
627    /// Marks an owner-reported cleanup wait as an elapsed deadline.
628    ///
629    /// This internal typed bit is deliberately separate from the public error
630    /// code: `ServiceOverloaded` is also used by ordinary admission and cannot
631    /// by itself authorize indefinite cleanup retries.
632    #[must_use]
633    #[doc(hidden)]
634    pub fn with_cleanup_deadline_elapsed(mut self) -> Self {
635        self.0.context.cleanup_deadline_elapsed = true;
636        self
637    }
638
639    #[doc(hidden)]
640    pub fn is_cleanup_deadline_elapsed(&self) -> bool {
641        self.0.context.cleanup_deadline_elapsed
642    }
643
644    /// Adds one machine-readable fact.
645    ///
646    /// `key` must be a non-empty static string and should be treated as part of
647    /// the diagnostic contract once exposed. `value` is stringified with
648    /// [`ToString`]. A duplicate or empty key never panics and never replaces
649    /// the primary error's facts. The first value wins, and the rejected key is
650    /// recorded under the `fact_collision_keys` fact. Layered `with_fact` calls must
651    /// therefore use distinct keys per layer; wrapping an error must not
652    /// re-emit a key the wrapped error already owns.
653    #[must_use]
654    pub fn with_fact(mut self, key: &'static str, value: impl ToString) -> Self {
655        self.0.facts.insert(key, value.to_string());
656        self
657    }
658
659    #[must_use]
660    #[doc(hidden)]
661    pub fn with_fact_if_absent(mut self, key: &'static str, value: impl ToString) -> Self {
662        if self.0.facts.get(key).is_none() && !self.0.context.projects(key) {
663            self.0.facts.insert(key, value.to_string());
664        }
665        self
666    }
667
668    /// Returns the typed failure evidence that decisions read.
669    pub fn context(&self) -> &FailureContext {
670        &self.0.context
671    }
672
673    /// Records the budget scope the failure passed through. The innermost
674    /// scope wins; a different later scope is reported as a fact collision.
675    #[must_use]
676    pub fn with_memory_failure(mut self, memory: MemoryFailure) -> Self {
677        let candidate = FailureContext { memory: Some(memory), ..FailureContext::default() };
678        match &self.0.context.memory {
679            None => self.0.context.memory = candidate.memory,
680            Some(_) => self.record_context_collisions(&candidate),
681        }
682        self
683    }
684
685    /// Records the execution boundary that owned the failed work. The first
686    /// complete boundary wins as a unit, so an outer scope never pairs its
687    /// node or side with an inner failure location.
688    #[must_use]
689    pub fn with_resource_boundary(mut self, boundary: ResourceBoundary) -> Self {
690        let candidate = FailureContext { boundary: Some(boundary), ..FailureContext::default() };
691        match &self.0.context.boundary {
692            None => self.0.context.boundary = candidate.boundary,
693            Some(_) => self.record_context_collisions(&candidate),
694        }
695        self
696    }
697
698    /// Records the native allocator exception family of an allocation failure.
699    #[must_use]
700    #[doc(hidden)]
701    pub fn with_allocation_source(mut self, source: AllocationSource) -> Self {
702        let candidate = FailureContext { allocation_source: Some(source), ..FailureContext::default() };
703        match self.0.context.allocation_source {
704            None => self.0.context.allocation_source = candidate.allocation_source,
705            Some(_) => self.record_context_collisions(&candidate),
706        }
707        self
708    }
709
710    /// Retains an exact high-water mark from a completed resident-graph cache
711    /// scope. It is typed evidence, never a wire fact, so consumers cannot
712    /// mistake diagnostic text for a peak.
713    #[must_use]
714    #[doc(hidden)]
715    pub fn with_graph_cache_scope_peak_bytes(mut self, peak_bytes: u64) -> Self {
716        self.0.context.graph_cache_scope_peak_bytes.get_or_insert(peak_bytes);
717        self
718    }
719
720    fn record_context_collisions(&mut self, candidate: &FailureContext) {
721        let current = self.0.context.projection();
722        for (key, value) in candidate.projection() {
723            if current.iter().any(|(existing, current)| *existing == key && *current != value) {
724                self.0.facts.record_collision(key);
725            }
726        }
727    }
728
729    /// Stored diagnostic facts plus the typed context rendered under its
730    /// wire keys, sorted by key.
731    ///
732    /// The typed value wins over a stored fact with the same key; a stored
733    /// fact that disagrees is reported under `fact_collision_keys`.
734    pub fn wire_facts(&self) -> impl Iterator<Item = (&'static str, Cow<'_, str>)> {
735        let mut facts = self
736            .0
737            .facts
738            .iter()
739            .filter(|(key, _)| *key != FACT_COLLISION_KEYS)
740            .map(|(key, value)| (key, Cow::Borrowed(value)))
741            .collect::<BTreeMap<_, _>>();
742        let mut collisions = self.0.facts.collision_keys.clone();
743        for (key, value) in self.0.context.projection() {
744            if facts.insert(key, value.clone()).is_some_and(|stored| stored != value) {
745                collisions.insert(key);
746            }
747        }
748        if !collisions.is_empty() {
749            facts.insert(FACT_COLLISION_KEYS, Cow::Owned(collisions.into_iter().collect::<Vec<_>>().join(",")));
750        }
751        facts.into_iter()
752    }
753
754    /// One value of [`Self::wire_facts`].
755    pub fn wire_fact(&self, key: &str) -> Option<Cow<'_, str>> {
756        self.wire_facts().find_map(|(candidate, value)| (candidate == key).then_some(value))
757    }
758
759    /// Assigns a query-wide diagnostic correlation id and mirrors it into facts.
760    #[must_use]
761    pub fn with_query_id(mut self, query_id: impl Into<QueryId>) -> Self {
762        let query_id = query_id.into();
763        self.0.facts.set("query_id", query_id.to_string());
764        self.0.query_id = Some(query_id);
765        self
766    }
767
768    /// Assigns an optional parent trace id and mirrors it into facts.
769    #[must_use]
770    pub fn with_trace_id(mut self, trace_id: impl Into<TraceId>) -> Self {
771        let trace_id = trace_id.into();
772        self.0.facts.set("trace_id", trace_id.to_string());
773        self.0.trace_id = Some(trace_id);
774        self
775    }
776
777    /// Attaches the source error when no source is already present.
778    ///
779    /// A second source is discarded so error decoration cannot replace the
780    /// original causal chain. Layers that need to add context wrap the error
781    /// in a new [`QueryError`] and do not re-attach a source.
782    #[must_use]
783    pub fn with_source<E>(mut self, source: E) -> Self
784    where E: StdError + Send + Sync + 'static {
785        if self.0.source.is_none() {
786            self.0.source = Some(Arc::new(source));
787        }
788        self
789    }
790
791    /// Returns the stable machine-readable error code.
792    pub fn code(&self) -> QueryErrorCode {
793        self.0.code
794    }
795
796    /// Returns the coarse category derived from [`Self::code`].
797    pub fn kind(&self) -> QueryErrorKind {
798        self.0.code.kind()
799    }
800
801    /// Returns the retry/remediation status derived from [`Self::code`].
802    pub fn status(&self) -> QueryErrorStatus {
803        self.0.code.status()
804    }
805
806    /// Returns the query-wide diagnostic correlation id, when assigned.
807    pub fn query_id(&self) -> Option<&QueryId> {
808        self.0.query_id.as_ref()
809    }
810
811    /// Returns the optional parent trace id, when assigned.
812    pub fn trace_id(&self) -> Option<&TraceId> {
813        self.0.trace_id.as_ref()
814    }
815
816    /// Returns the stable dotted operation id that produced the error.
817    pub fn operation(&self) -> &'static str {
818        self.0.operation
819    }
820
821    /// Returns the human-facing diagnostic message.
822    pub fn message(&self) -> &str {
823        &self.0.message
824    }
825
826    /// Returns the sorted machine-readable fact set.
827    pub fn facts(&self) -> &Facts {
828        self.0.facts.as_ref()
829    }
830
831    /// Returns a stored fact by name. Typed failure-context projections are
832    /// available through [`Self::wire_facts`].
833    pub fn fact_value(&self, key: &str) -> Option<&str> {
834        self.0.facts.get(key)
835    }
836
837    /// Returns the underlying source error when one was attached.
838    pub fn source_error(&self) -> Option<&(dyn StdError + 'static)> {
839        self.0.source.as_deref().map(|source| source as &(dyn StdError + 'static))
840    }
841
842    /// Serializes this error as a structured diagnostic JSON object.
843    pub fn to_json(&self) -> Value {
844        let query_id = self.0.query_id.as_ref().map(QueryId::as_str).or_else(|| self.fact_value("query_id"));
845        let trace_id = self.0.trace_id.as_ref().map(TraceId::as_str).or_else(|| self.fact_value("trace_id"));
846        json!({
847            "schema": "query_engine.query_error",
848            "query_id": query_id,
849            "trace_id": trace_id,
850            "triage": self.triage_summary().to_json(),
851            "code": self.0.code.as_str(),
852            "kind": self.kind().as_str(),
853            "status": self.status().as_str(),
854            "operation": self.0.operation,
855            "message": self.0.message,
856            "facts": self.wire_facts().collect::<BTreeMap<_, _>>(),
857            "source": self.0.source.as_ref().map(|source| source.to_string()),
858        })
859    }
860}
861
862/// Redacts whitespace-delimited `scheme://...` tokens and normalizes spaces.
863///
864/// Native boundaries project foreign error text into diagnostic facts. They
865/// share this helper so one redaction policy covers every such projection,
866/// whichever component produced the message.
867pub fn redact_uri_tokens(message: &str) -> String {
868    message
869        .split_whitespace()
870        .map(|token| if token.contains("://") { "<redacted-uri>" } else { token })
871        .collect::<Vec<_>>()
872        .join(" ")
873}
874
875/// Renders a byte count as exact bytes followed by GiB truncated to two
876/// decimals, such as `101973819392 bytes (94.97 GiB)`.
877///
878/// Capacity and cap messages share this form so the exact value that must be
879/// compared with a limit and the figure an operator recognizes appear
880/// together. Truncation keeps a limit from reading larger than it is.
881pub fn bytes_with_gib(bytes: u64) -> String {
882    const GIB: u128 = 1 << 30;
883    let hundredths = u128::from(bytes) * 100 / GIB;
884    format!("{bytes} bytes ({}.{:02} GiB)", hundredths / 100, hundredths % 100)
885}
886
887impl fmt::Display for QueryError {
888    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
889        write!(
890            f,
891            "code={} kind={} status={} operation={} message={}",
892            self.code(),
893            self.kind(),
894            self.status(),
895            self.operation(),
896            quote(self.message()),
897        )?;
898
899        f.write_str(" facts={")?;
900        for (index, (key, value)) in self.wire_facts().enumerate() {
901            if index > 0 {
902                f.write_str(", ")?;
903            }
904            write!(f, "{key}={}", quote(&value))?;
905        }
906        f.write_str("}")?;
907
908        if let Some(source) = &self.0.source {
909            write!(f, " source={}", quote(&source.to_string()))?;
910        }
911
912        Ok(())
913    }
914}
915
916impl fmt::Debug for QueryError {
917    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
918        f.debug_struct("QueryError")
919            .field("query_id", &self.0.query_id)
920            .field("trace_id", &self.0.trace_id)
921            .field("code", &self.0.code)
922            .field("kind", &self.kind())
923            .field("status", &self.status())
924            .field("operation", &self.0.operation)
925            .field("message", &self.0.message)
926            .field("facts", &self.0.facts)
927            .field("context", &self.0.context)
928            .field("source", &self.0.source.as_ref().map(|source| source.to_string()))
929            .finish()
930    }
931}
932
933impl StdError for QueryError {
934    fn source(&self) -> Option<&(dyn StdError + 'static)> {
935        self.source_error()
936    }
937}
938
939fn quote(value: &str) -> String {
940    if value.chars().all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '_' | '-' | '.' | ':' | '/' | '(' | ')')) {
941        value.to_string()
942    } else {
943        format!("{value:?}")
944    }
945}
946
947#[cfg(test)]
948mod tests {
949    use super::*;
950    use crate::plan::NodeId;
951
952    #[test]
953    fn bytes_with_gib_prints_exact_bytes_and_truncated_gib() {
954        assert_eq!(bytes_with_gib(0), "0 bytes (0.00 GiB)");
955        assert_eq!(bytes_with_gib(700), "700 bytes (0.00 GiB)");
956        assert_eq!(bytes_with_gib(95 * (1 << 30)), "102005473280 bytes (95.00 GiB)");
957        assert_eq!(bytes_with_gib(101_973_819_392), "101973819392 bytes (94.97 GiB)");
958        assert_eq!(bytes_with_gib(95 * (1 << 30) - 1), "102005473279 bytes (94.99 GiB)");
959        assert_eq!(bytes_with_gib(u64::MAX), "18446744073709551615 bytes (17179869183.99 GiB)");
960    }
961
962    #[test]
963    fn redact_uri_tokens_replaces_uri_like_tokens_only() {
964        assert_eq!(
965            redact_uri_tokens("S3: must provide `aws_access_key` if AWS_ACCESS_KEY_ID isn't set."),
966            "S3: must provide `aws_access_key` if AWS_ACCESS_KEY_ID isn't set."
967        );
968        assert_eq!(
969            redact_uri_tokens("failed to open s3://bucket/secret/key.parquet for read"),
970            "failed to open <redacted-uri> for read"
971        );
972        assert_eq!(redact_uri_tokens("  spaced   message  "), "spaced message");
973    }
974
975    #[test]
976    fn graph_cache_scope_peak_is_a_typed_sideband_not_an_error_fact() {
977        let error = QueryError::new(
978            QueryErrorCode::DeviceAllocationFailed,
979            "query_engine.test.cache_scope",
980            "cache scope failed",
981        )
982        .with_graph_cache_scope_peak_bytes(4_096);
983
984        assert_eq!(error.context().graph_cache_scope_peak_bytes, Some(4_096));
985        assert_eq!(error.fact_value("graph_cache_scope_peak_bytes"), None);
986        assert!(error.to_json()["facts"].get("graph_cache_scope_peak_bytes").is_none());
987    }
988
989    #[test]
990    fn wire_facts_render_typed_context_over_disagreeing_stored_facts() {
991        let node = NodeId::from_index_for_test(4);
992        let error = QueryError::new(
993            QueryErrorCode::QueryMemoryLimitExceeded,
994            "query_engine.test.wire_facts",
995            "typed context projection probe",
996        )
997        .with_fact("budget_bytes", 1)
998        .with_fact("node_id", node)
999        .with_memory_failure(MemoryFailure {
1000            boundary: "inner".into(),
1001            operation: "cudf.test.op".into(),
1002            usage: Some(MemoryUsage {
1003                limit_bytes: 4096,
1004                peak_bytes: 3,
1005                peak_aligned_bytes: 256,
1006                current_bytes: 2,
1007                current_aligned_bytes: 256,
1008                failed_allocation_count: 1,
1009                largest_failed_allocation_bytes: 8192,
1010            }),
1011        })
1012        .with_memory_failure(MemoryFailure { boundary: "outer".into(), operation: "cudf.test.op".into(), usage: None })
1013        .with_resource_boundary(ResourceBoundary::new("source_chunk_read", node).with_resume("source_chunk"))
1014        .with_resource_boundary(ResourceBoundary::new("outer_stage", node))
1015        .with_fact_if_absent("resume_boundary", "outer_resume");
1016
1017        let wire = error.wire_facts().map(|(key, value)| (key, value.into_owned())).collect::<BTreeMap<_, _>>();
1018        assert_eq!(wire["budget_bytes"], "4096", "the allocator limit wins over a stale stored fact");
1019        assert_eq!(wire["budget_boundary"], "inner", "the innermost scope wins");
1020        assert_eq!(wire["memory_boundary"], "source_chunk_read", "the first boundary wins as a unit");
1021        assert_eq!(wire["resume_boundary"], "source_chunk", "a typed key counts as present");
1022        assert_eq!(wire["node_id"], "n4");
1023        assert_eq!(wire[FACT_COLLISION_KEYS], "budget_boundary,budget_bytes,memory_boundary,stage");
1024        assert_eq!(error.wire_facts().collect::<Vec<_>>(), error.wire_facts().collect::<Vec<_>>());
1025        assert!(error.to_string().contains("budget_bytes=4096"), "{error}");
1026    }
1027
1028    #[test]
1029    fn attempt_failure_codes_round_trip_through_their_labels() {
1030        let codes = QueryErrorCode::ALL
1031            .iter()
1032            .map(|code| AttemptFailureCode::Query(*code))
1033            .chain(AttemptFailureCode::LIFECYCLE)
1034            .collect::<Vec<_>>();
1035        for code in codes {
1036            let json = serde_json::to_string(&code).unwrap();
1037            assert_eq!(json, format!("\"{}\"", code.as_str()));
1038            assert_eq!(serde_json::from_str::<AttemptFailureCode>(&json).unwrap(), code);
1039        }
1040        assert_eq!(AttemptFailureCode::Query(QueryErrorCode::Internal).as_str(), "internal");
1041        assert!(serde_json::from_str::<AttemptFailureCode>("\"unsupported_plan_shape\"").is_err());
1042    }
1043
1044    #[test]
1045    fn invalid_configuration_is_a_permanent_invalid_input_code_distinct_from_invalid_plan() {
1046        let code = QueryErrorCode::InvalidConfiguration;
1047        assert_eq!(code.as_str(), "invalid_configuration");
1048        assert_eq!(code.kind(), QueryErrorKind::InvalidInput);
1049        assert_eq!(code.status(), QueryErrorStatus::Permanent);
1050        assert_eq!(TriageCode::from(code), TriageCode::InvalidInput);
1051        assert_eq!(QueryErrorCode::InvalidPlan.kind(), QueryErrorKind::Planning);
1052        assert!(QueryErrorCode::ALL.contains(&code));
1053    }
1054
1055    #[test]
1056    fn duplicate_fact_decoration_preserves_first_value_and_records_collision() {
1057        let error = QueryError::new(
1058            QueryErrorCode::CudfInvalidInput,
1059            "query_engine.test.fact_collision",
1060            "fact collision probe",
1061        )
1062        .with_fact("layer", "outer")
1063        .with_fact("layer", "inner")
1064        .with_fact("", "empty key is ignored");
1065
1066        assert_eq!(error.fact_value("layer"), Some("outer"));
1067        assert_eq!(error.fact_value(FACT_COLLISION_KEYS), Some("<empty>,layer"));
1068        assert_eq!(error.code(), QueryErrorCode::CudfInvalidInput);
1069    }
1070
1071    #[test]
1072    fn repeated_source_decoration_preserves_first_cause() {
1073        let first =
1074            QueryError::new(QueryErrorCode::CudfInvalidInput, "query_engine.test.first_cause", "the original cause");
1075        let expected_cause = first.to_string();
1076        let error = QueryError::new(
1077            QueryErrorCode::CudfInvalidInput,
1078            "query_engine.test.source_preservation",
1079            "source preservation probe",
1080        )
1081        .with_source(first)
1082        .with_source(QueryError::new(
1083            QueryErrorCode::DeviceFault,
1084            "query_engine.test.decoration",
1085            "decoration must not replace the cause",
1086        ));
1087
1088        let cause = error.source_error().expect("first source must be retained").to_string();
1089        assert_eq!(cause, expected_cause, "{cause}");
1090        assert!(!cause.contains("decoration must not replace the cause"), "{cause}");
1091    }
1092}
query_planning/error.rs

query_planning/
error.rs
