algeon-query-runtime 0.1.2

Native query runtime, lifecycle, ledger, and observation owners.
Documentation 
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

use query_planning::error::QueryError;

/// Retry/resume facts attached to bounded retry failures.
///
/// These fields document what bounded execution can safely retry after a
/// retryable resource-exhaustion failure: the retry action, resume boundary,
/// already-committed chunks/partials, and merge fan-in. The fact keys emitted
/// by [`Self::attach_to_error`] are stable diagnostics used by tests and higher
/// level error reports.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RetryFacts {
    /// Stable retry action string, for example splitting a chunk or reducing
    /// merge fan-in.
    pub(crate) retry_action: Option<String>,
    /// Stable boundary where execution can resume after a retry.
    pub(crate) resume_boundary: Option<String>,
    /// Number of source chunks already committed before the failure.
    pub(crate) committed_chunks: Option<usize>,
    /// Number of partial results already committed before the failure.
    pub(crate) committed_partials: Option<usize>,
    /// Merge level being processed when the failure occurred.
    pub(crate) merge_level: Option<usize>,
    /// Merge fan-in attempted when the failure occurred.
    pub(crate) merge_fan_in: Option<usize>,
}

impl RetryFacts {
    /// Creates empty retry facts.
    #[doc(hidden)]
    pub fn new() -> Self {
        Self::default()
    }

    /// Sets the stable retry-action diagnostic value.
    #[must_use]
    #[doc(hidden)]
    pub fn with_retry_action(mut self, retry_action: impl Into<String>) -> Self {
        self.retry_action = Some(retry_action.into());
        self
    }

    /// Sets the stable resume-boundary diagnostic value.
    #[must_use]
    #[doc(hidden)]
    pub fn with_resume_boundary(mut self, resume_boundary: impl Into<String>) -> Self {
        self.resume_boundary = Some(resume_boundary.into());
        self
    }

    /// Records committed source chunks.
    #[must_use]
    #[doc(hidden)]
    pub fn with_committed_chunks(mut self, committed_chunks: usize) -> Self {
        self.committed_chunks = Some(committed_chunks);
        self
    }

    /// Records committed partial results.
    #[must_use]
    #[doc(hidden)]
    pub fn with_committed_partials(mut self, committed_partials: usize) -> Self {
        self.committed_partials = Some(committed_partials);
        self
    }

    /// Records the merge level involved in a retryable failure.
    #[must_use]
    #[doc(hidden)]
    pub fn with_merge_level(mut self, merge_level: usize) -> Self {
        self.merge_level = Some(merge_level);
        self
    }

    /// Records the merge fan-in involved in a retryable failure.
    #[must_use]
    #[doc(hidden)]
    pub fn with_merge_fan_in(mut self, merge_fan_in: usize) -> Self {
        self.merge_fan_in = Some(merge_fan_in);
        self
    }

    /// Attaches all present retry facts to an error.
    ///
    /// This preserves the original error code/status and only adds structured
    /// facts. Fact keys are stable diagnostic contracts.
    #[doc(hidden)]
    pub fn attach_to_error(&self, mut error: QueryError) -> QueryError {
        if let Some(retry_action) = &self.retry_action {
            error = error.with_fact_if_absent("retry_action", retry_action);
        }
        if let Some(resume_boundary) = &self.resume_boundary {
            error = error.with_fact_if_absent("resume_boundary", resume_boundary);
        }
        if let Some(committed_chunks) = self.committed_chunks {
            error = error.with_fact_if_absent("committed_chunks", committed_chunks);
        }
        if let Some(committed_partials) = self.committed_partials {
            error = error.with_fact_if_absent("committed_partials", committed_partials);
        }
        if let Some(merge_level) = self.merge_level {
            error = error.with_fact_if_absent("merge_level", merge_level);
        }
        if let Some(merge_fan_in) = self.merge_fan_in {
            error = error.with_fact_if_absent("merge_fan_in", merge_fan_in);
        }
        error
    }
}

#[cfg(test)]
mod tests {
    use query_planning::error::{QueryError, QueryErrorCode};

    use super::*;

    #[test]
    fn retry_facts_attach_stable_resume_and_merge_context() {
        let error = RetryFacts::new()
            .with_retry_action("reduce_merge_fan_in")
            .with_resume_boundary("final_merge")
            .with_committed_chunks(8)
            .with_committed_partials(8)
            .with_merge_level(1)
            .with_merge_fan_in(2)
            .attach_to_error(QueryError::new(
                QueryErrorCode::DeviceAllocationFailed,
                "query_engine.test",
                "allocation failed",
            ));

        assert_eq!(error.fact_value("retry_action"), Some("reduce_merge_fan_in"));
        assert_eq!(error.fact_value("resume_boundary"), Some("final_merge"));
        assert_eq!(error.fact_value("committed_chunks"), Some("8"));
        assert_eq!(error.fact_value("committed_partials"), Some("8"));
        assert_eq!(error.fact_value("merge_level"), Some("1"));
        assert_eq!(error.fact_value("merge_fan_in"), Some("2"));
    }
}