dataprof_core/
execution.rs

1/// A failed execution attempt followed by a retry, in chronological order.
2///
3/// This is execution provenance, not a metric or a count of damaged rows.
4/// Error text is diagnostic and is not a stable machine-readable contract.
5#[derive(
6    Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, schemars::JsonSchema,
7)]
8pub struct RecoveryEvent {
9    pub kind: RecoveryKind,
10    /// Engine or parse strategy that failed.
11    pub attempted: String,
12    /// Engine or parse strategy tried next.
13    pub retry: String,
14    pub error: String,
15}
16
17/// Recovery boundaries recorded by dataprof.
18#[derive(
19    Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, schemars::JsonSchema,
20)]
21#[serde(rename_all = "snake_case")]
22pub enum RecoveryKind {
23    EngineFallback,
24    CsvAutoRecovery,
25}
26
27impl RecoveryKind {
28    /// Stable serialized name, also used by the Python binding.
29    pub fn as_str(self) -> &'static str {
30        match self {
31            Self::EngineFallback => "engine_fallback",
32            Self::CsvAutoRecovery => "csv_auto_recovery",
33        }
34    }
35}
36
37/// Reason why profiling was truncated before exhausting the source.
38#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize, schemars::JsonSchema)]
39pub enum TruncationReason {
40    /// Stopped after processing a maximum number of rows.
41    MaxRows(u64),
42    /// Stopped after consuming a maximum number of bytes.
43    MaxBytes(u64),
44    /// Stopped due to memory pressure.
45    MemoryPressure,
46    /// Stopped due to a user-defined stop condition.
47    StopCondition(String),
48    /// The input stream was closed by the producer.
49    StreamClosed,
50    /// Stopped due to a timeout.
51    Timeout,
52}
53
54/// Metadata about the profiling execution.
55#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, schemars::JsonSchema)]
56pub struct ExecutionMetadata {
57    /// Engine or parser that actually produced the report.
58    #[serde(default, skip_serializing_if = "Option::is_none")]
59    pub engine: Option<String>,
60    /// Failed attempts and their retries. None means history was not recorded
61    /// (e.g. a legacy report); Some([]) means no recovery occurred.
62    #[serde(default, skip_serializing_if = "Option::is_none")]
63    pub recovery_events: Option<Vec<RecoveryEvent>>,
64    /// Number of rows actually processed or analyzed.
65    pub rows_processed: usize,
66    /// Number of bytes consumed from the source, if known.
67    #[serde(skip_serializing_if = "Option::is_none")]
68    pub bytes_consumed: Option<u64>,
69    /// Number of columns detected in the data.
70    pub columns_detected: usize,
71    /// Total execution time in milliseconds.
72    pub scan_time_ms: u128,
73    /// Throughput in rows per second, auto-calculated when possible.
74    #[serde(skip_serializing_if = "Option::is_none")]
75    pub throughput_rows_sec: Option<f64>,
76    /// Peak process memory usage (resident set size) in megabytes observed
77    /// during profiling, sampled at chunk/batch boundaries. Absent when the
78    /// platform provided no reading or the input path does not track memory.
79    #[serde(skip_serializing_if = "Option::is_none")]
80    pub memory_peak_mb: Option<f64>,
81    /// Number of errors encountered during profiling.
82    pub error_count: usize,
83    /// Number of data rows whose field count differed from the header's.
84    ///
85    /// These rows were recovered — extra trailing fields dropped, missing
86    /// trailing fields padded to null — so profiling continued, but each one
87    /// is a structural violation in the source. A nonzero value means the file
88    /// did not parse cleanly even though a report was produced. Additive field:
89    /// legacy documents without it deserialize as `0`.
90    #[serde(default)]
91    #[schemars(default)]
92    pub ragged_row_count: usize,
93    /// Whether a CSV source ended inside a quoted field.
94    ///
95    /// A quote that is never closed swallows every following row into the
96    /// last field of one record, and the parser accepts it. `true` means the
97    /// last record read may hold several source rows. Absent where it was not
98    /// checked: non-CSV input, a scan that stopped before the end of the
99    /// source, and reports written before 0.12.
100    #[serde(default, skip_serializing_if = "Option::is_none")]
101    pub unterminated_quote: Option<bool>,
102    /// Whether the entire source was consumed.
103    pub source_exhausted: bool,
104    /// If the source was not exhausted, why processing stopped.
105    #[serde(skip_serializing_if = "Option::is_none")]
106    pub truncation_reason: Option<TruncationReason>,
107    /// Whether sampling was applied.
108    pub sampling_applied: bool,
109    /// Ratio of rows analyzed to total rows when sampling is meaningful.
110    #[serde(skip_serializing_if = "Option::is_none")]
111    pub sampling_ratio: Option<f64>,
112    /// Exact zero-based, half-open source row ranges selected for analysis,
113    /// in source order. Absent when the input path does not record a selection;
114    /// an empty list means a recorded selection of zero rows.
115    #[serde(default, skip_serializing_if = "Option::is_none")]
116    pub sampled_row_ranges: Option<Vec<[u64; 2]>>,
117}
118
119impl ExecutionMetadata {
120    /// Create new execution metadata with throughput calculated automatically.
121    pub fn new(rows_processed: usize, columns_detected: usize, scan_time_ms: u128) -> Self {
122        let throughput_rows_sec = if scan_time_ms > 0 {
123            Some(rows_processed as f64 / (scan_time_ms as f64 / 1000.0))
124        } else {
125            None
126        };
127
128        Self {
129            engine: None,
130            recovery_events: Some(Vec::new()),
131            rows_processed,
132            bytes_consumed: None,
133            columns_detected,
134            scan_time_ms,
135            throughput_rows_sec,
136            memory_peak_mb: None,
137            error_count: 0,
138            ragged_row_count: 0,
139            unterminated_quote: None,
140            source_exhausted: true,
141            truncation_reason: None,
142            sampling_applied: false,
143            sampling_ratio: None,
144            sampled_row_ranges: None,
145        }
146    }
147
148    /// Record the engine or parser that actually produced the report.
149    pub fn with_engine(mut self, engine: impl Into<String>) -> Self {
150        self.engine = Some(engine.into());
151        self
152    }
153
154    /// Set sampling information.
155    pub fn with_sampling(mut self, ratio: f64) -> Self {
156        self.sampling_applied = true;
157        self.sampling_ratio = Some(ratio);
158        self
159    }
160
161    /// Explicitly set whether the source was fully consumed.
162    pub fn with_source_exhausted(mut self, exhausted: bool) -> Self {
163        self.source_exhausted = exhausted;
164        self
165    }
166
167    /// Mark the execution as truncated.
168    pub fn with_truncation(mut self, reason: TruncationReason) -> Self {
169        self.source_exhausted = false;
170        self.truncation_reason = Some(reason);
171        self
172    }
173
174    /// Set the number of bytes consumed from the source.
175    pub fn with_bytes_consumed(mut self, bytes: u64) -> Self {
176        self.bytes_consumed = Some(bytes);
177        self
178    }
179
180    /// Set the error count.
181    pub fn with_error_count(mut self, count: usize) -> Self {
182        self.error_count = count;
183        self
184    }
185
186    /// Set the count of ragged rows (field count != header field count).
187    pub fn with_ragged_row_count(mut self, count: usize) -> Self {
188        self.ragged_row_count = count;
189        self
190    }
191
192    /// Record whether a checked CSV source ended inside a quoted field.
193    pub fn with_unterminated_quote(mut self, ended_inside_quotes: bool) -> Self {
194        self.unterminated_quote = Some(ended_inside_quotes);
195        self
196    }
197
198    /// Set peak memory usage.
199    pub fn with_memory_peak_mb(mut self, mb: f64) -> Self {
200        self.memory_peak_mb = Some(mb);
201        self
202    }
203}
204
205#[cfg(test)]
206mod tests {
207    use super::*;
208
209    #[test]
210    fn test_execution_metadata_throughput_calculation() {
211        let meta = ExecutionMetadata::new(1000, 5, 500);
212        assert!(meta.throughput_rows_sec.is_some());
213        assert!((meta.throughput_rows_sec.unwrap() - 2000.0).abs() < 1.0);
214        assert!(meta.source_exhausted);
215        assert!(!meta.sampling_applied);
216        assert!(meta.sampling_ratio.is_none());
217    }
218
219    #[test]
220    fn test_execution_metadata_zero_time_no_throughput() {
221        let meta = ExecutionMetadata::new(100, 3, 0);
222        assert!(meta.throughput_rows_sec.is_none());
223    }
224
225    #[test]
226    fn test_execution_metadata_with_engine() {
227        let meta = ExecutionMetadata::new(100, 3, 25).with_engine("incremental");
228        assert_eq!(meta.engine.as_deref(), Some("incremental"));
229    }
230
231    #[test]
232    fn test_execution_metadata_with_sampling() {
233        let meta = ExecutionMetadata::new(500, 3, 100).with_sampling(0.5);
234        assert!(meta.sampling_applied);
235        assert_eq!(meta.sampling_ratio, Some(0.5));
236    }
237
238    #[test]
239    fn test_execution_metadata_ragged_row_count() {
240        let meta = ExecutionMetadata::new(100, 3, 25);
241        assert_eq!(meta.ragged_row_count, 0);
242        let meta = meta.with_ragged_row_count(2);
243        assert_eq!(meta.ragged_row_count, 2);
244    }
245
246    #[test]
247    fn test_execution_metadata_ragged_defaults_when_absent() {
248        // A document written before the field existed must deserialize as 0,
249        // not fail or fabricate a value.
250        let json = r#"{
251            "rows_processed": 10,
252            "columns_detected": 2,
253            "scan_time_ms": 5,
254            "error_count": 0,
255            "source_exhausted": true,
256            "sampling_applied": false
257        }"#;
258        let meta: ExecutionMetadata = serde_json::from_str(json).unwrap();
259        assert_eq!(meta.ragged_row_count, 0);
260        assert_eq!(meta.sampled_row_ranges, None);
261        assert_eq!(meta.recovery_events, None);
262        assert!(
263            serde_json::to_value(&meta)
264                .unwrap()
265                .get("recovery_events")
266                .is_none()
267        );
268        assert!(
269            serde_json::to_value(&meta)
270                .unwrap()
271                .get("sampled_row_ranges")
272                .is_none()
273        );
274    }
275
276    #[test]
277    fn recorded_row_selection_preserves_empty_and_nonempty_ranges() {
278        for ranges in [vec![], vec![[0, 1], [99, 100]]] {
279            let mut meta = ExecutionMetadata::new(0, 1, 0);
280            meta.sampled_row_ranges = Some(ranges.clone());
281            let json = serde_json::to_value(&meta).unwrap();
282            let restored: ExecutionMetadata = serde_json::from_value(json).unwrap();
283            assert_eq!(restored.sampled_row_ranges, Some(ranges));
284        }
285    }
286
287    #[test]
288    fn recovery_history_preserves_absence_empty_and_ordered_events() {
289        let events = vec![
290            RecoveryEvent {
291                kind: RecoveryKind::EngineFallback,
292                attempted: "columnar".into(),
293                retry: "incremental".into(),
294                error: "primary failed".into(),
295            },
296            RecoveryEvent {
297                kind: RecoveryKind::CsvAutoRecovery,
298                attempted: "strict".into(),
299                retry: "flexible".into(),
300                error: "unequal field count".into(),
301            },
302        ];
303        assert_eq!(
304            ExecutionMetadata::new(0, 0, 0).recovery_events,
305            Some(vec![])
306        );
307        for history in [None, Some(vec![]), Some(events)] {
308            let mut meta = ExecutionMetadata::new(0, 0, 0);
309            meta.recovery_events = history.clone();
310            let json = serde_json::to_value(&meta).unwrap();
311            if let Some(events) = &history {
312                for (index, event) in events.iter().enumerate() {
313                    assert_eq!(json["recovery_events"][index]["kind"], event.kind.as_str());
314                }
315            }
316            let restored: ExecutionMetadata = serde_json::from_value(json).unwrap();
317            assert_eq!(restored.recovery_events, history);
318        }
319    }
320
321    #[test]
322    fn test_execution_metadata_with_truncation() {
323        let meta =
324            ExecutionMetadata::new(1000, 5, 200).with_truncation(TruncationReason::MaxRows(1000));
325        assert!(!meta.source_exhausted);
326        assert!(meta.truncation_reason.is_some());
327    }
328
329    #[test]
330    fn test_truncation_reason_serde_roundtrip() {
331        let reasons = vec![
332            TruncationReason::MaxRows(5000),
333            TruncationReason::MaxBytes(1_000_000),
334            TruncationReason::MemoryPressure,
335            TruncationReason::StopCondition("accuracy > 0.95".to_string()),
336            TruncationReason::StreamClosed,
337            TruncationReason::Timeout,
338        ];
339
340        for reason in reasons {
341            let json = serde_json::to_string(&reason).unwrap();
342            let deserialized: TruncationReason = serde_json::from_str(&json).unwrap();
343            let json2 = serde_json::to_string(&deserialized).unwrap();
344            assert_eq!(json, json2);
345        }
346    }
347}
dataprof_core/execution.rs

dataprof_core/
execution.rs
