dataprof-core 0.13.0

/// A failed execution attempt followed by a retry, in chronological order.
///
/// This is execution provenance, not a metric or a count of damaged rows.
/// Error text is diagnostic and is not a stable machine-readable contract.
#[derive(
    Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, schemars::JsonSchema,
)]
pub struct RecoveryEvent {
    pub kind: RecoveryKind,
    /// Engine or parse strategy that failed.
    pub attempted: String,
    /// Engine or parse strategy tried next.
    pub retry: String,
    pub error: String,
}

/// Recovery boundaries recorded by dataprof.
#[derive(
    Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, schemars::JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum RecoveryKind {
    EngineFallback,
    CsvAutoRecovery,
}

impl RecoveryKind {
    /// Stable serialized name, also used by the Python binding.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::EngineFallback => "engine_fallback",
            Self::CsvAutoRecovery => "csv_auto_recovery",
        }
    }
}

/// Reason why profiling was truncated before exhausting the source.
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize, schemars::JsonSchema)]
pub enum TruncationReason {
    /// Stopped after processing a maximum number of rows.
    MaxRows(u64),
    /// Stopped after consuming a maximum number of bytes.
    MaxBytes(u64),
    /// Stopped due to memory pressure.
    MemoryPressure,
    /// Stopped due to a user-defined stop condition.
    StopCondition(String),
    /// The input stream was closed by the producer.
    StreamClosed,
    /// Stopped due to a timeout.
    Timeout,
}

/// Metadata about the profiling execution.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, schemars::JsonSchema)]
pub struct ExecutionMetadata {
    /// Engine or parser that actually produced the report.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub engine: Option<String>,
    /// Failed attempts and their retries. None means history was not recorded
    /// (e.g. a legacy report); Some([]) means no recovery occurred.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub recovery_events: Option<Vec<RecoveryEvent>>,
    /// Number of rows actually processed or analyzed.
    pub rows_processed: usize,
    /// Number of bytes consumed from the source, if known.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub bytes_consumed: Option<u64>,
    /// Number of columns detected in the data.
    pub columns_detected: usize,
    /// Total execution time in milliseconds.
    pub scan_time_ms: u128,
    /// Throughput in rows per second, auto-calculated when possible.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub throughput_rows_sec: Option<f64>,
    /// Peak process memory usage (resident set size) in megabytes observed
    /// during profiling, sampled at chunk/batch boundaries. Absent when the
    /// platform provided no reading or the input path does not track memory.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub memory_peak_mb: Option<f64>,
    /// Number of errors encountered during profiling.
    pub error_count: usize,
    /// Number of data rows whose field count differed from the header's.
    ///
    /// These rows were recovered — extra trailing fields dropped, missing
    /// trailing fields padded to null — so profiling continued, but each one
    /// is a structural violation in the source. A nonzero value means the file
    /// did not parse cleanly even though a report was produced. Additive field:
    /// legacy documents without it deserialize as `0`.
    #[serde(default)]
    #[schemars(default)]
    pub ragged_row_count: usize,
    /// Whether a CSV source ended inside a quoted field.
    ///
    /// A quote that is never closed swallows every following row into the
    /// last field of one record, and the parser accepts it. `true` means the
    /// last record read may hold several source rows. Absent where it was not
    /// checked: non-CSV input, a scan that stopped before the end of the
    /// source, and reports written before 0.12.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub unterminated_quote: Option<bool>,
    /// Whether the entire source was consumed.
    pub source_exhausted: bool,
    /// If the source was not exhausted, why processing stopped.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub truncation_reason: Option<TruncationReason>,
    /// Whether sampling was applied.
    pub sampling_applied: bool,
    /// Ratio of rows analyzed to total rows when sampling is meaningful.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub sampling_ratio: Option<f64>,
    /// Exact zero-based, half-open source row ranges selected for analysis,
    /// in source order. Absent when the input path does not record a selection;
    /// an empty list means a recorded selection of zero rows.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub sampled_row_ranges: Option<Vec<[u64; 2]>>,
}

impl ExecutionMetadata {
    /// Create new execution metadata with throughput calculated automatically.
    pub fn new(rows_processed: usize, columns_detected: usize, scan_time_ms: u128) -> Self {
        let throughput_rows_sec = if scan_time_ms > 0 {
            Some(rows_processed as f64 / (scan_time_ms as f64 / 1000.0))
        } else {
            None
        };

        Self {
            engine: None,
            recovery_events: Some(Vec::new()),
            rows_processed,
            bytes_consumed: None,
            columns_detected,
            scan_time_ms,
            throughput_rows_sec,
            memory_peak_mb: None,
            error_count: 0,
            ragged_row_count: 0,
            unterminated_quote: None,
            source_exhausted: true,
            truncation_reason: None,
            sampling_applied: false,
            sampling_ratio: None,
            sampled_row_ranges: None,
        }
    }

    /// Record the engine or parser that actually produced the report.
    pub fn with_engine(mut self, engine: impl Into<String>) -> Self {
        self.engine = Some(engine.into());
        self
    }

    /// Set sampling information.
    pub fn with_sampling(mut self, ratio: f64) -> Self {
        self.sampling_applied = true;
        self.sampling_ratio = Some(ratio);
        self
    }

    /// Explicitly set whether the source was fully consumed.
    pub fn with_source_exhausted(mut self, exhausted: bool) -> Self {
        self.source_exhausted = exhausted;
        self
    }

    /// Mark the execution as truncated.
    pub fn with_truncation(mut self, reason: TruncationReason) -> Self {
        self.source_exhausted = false;
        self.truncation_reason = Some(reason);
        self
    }

    /// Set the number of bytes consumed from the source.
    pub fn with_bytes_consumed(mut self, bytes: u64) -> Self {
        self.bytes_consumed = Some(bytes);
        self
    }

    /// Set the error count.
    pub fn with_error_count(mut self, count: usize) -> Self {
        self.error_count = count;
        self
    }

    /// Set the count of ragged rows (field count != header field count).
    pub fn with_ragged_row_count(mut self, count: usize) -> Self {
        self.ragged_row_count = count;
        self
    }

    /// Record whether a checked CSV source ended inside a quoted field.
    pub fn with_unterminated_quote(mut self, ended_inside_quotes: bool) -> Self {
        self.unterminated_quote = Some(ended_inside_quotes);
        self
    }

    /// Set peak memory usage.
    pub fn with_memory_peak_mb(mut self, mb: f64) -> Self {
        self.memory_peak_mb = Some(mb);
        self
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_execution_metadata_throughput_calculation() {
        let meta = ExecutionMetadata::new(1000, 5, 500);
        assert!(meta.throughput_rows_sec.is_some());
        assert!((meta.throughput_rows_sec.unwrap() - 2000.0).abs() < 1.0);
        assert!(meta.source_exhausted);
        assert!(!meta.sampling_applied);
        assert!(meta.sampling_ratio.is_none());
    }

    #[test]
    fn test_execution_metadata_zero_time_no_throughput() {
        let meta = ExecutionMetadata::new(100, 3, 0);
        assert!(meta.throughput_rows_sec.is_none());
    }

    #[test]
    fn test_execution_metadata_with_engine() {
        let meta = ExecutionMetadata::new(100, 3, 25).with_engine("incremental");
        assert_eq!(meta.engine.as_deref(), Some("incremental"));
    }

    #[test]
    fn test_execution_metadata_with_sampling() {
        let meta = ExecutionMetadata::new(500, 3, 100).with_sampling(0.5);
        assert!(meta.sampling_applied);
        assert_eq!(meta.sampling_ratio, Some(0.5));
    }

    #[test]
    fn test_execution_metadata_ragged_row_count() {
        let meta = ExecutionMetadata::new(100, 3, 25);
        assert_eq!(meta.ragged_row_count, 0);
        let meta = meta.with_ragged_row_count(2);
        assert_eq!(meta.ragged_row_count, 2);
    }

    #[test]
    fn test_execution_metadata_ragged_defaults_when_absent() {
        // A document written before the field existed must deserialize as 0,
        // not fail or fabricate a value.
        let json = r#"{
            "rows_processed": 10,
            "columns_detected": 2,
            "scan_time_ms": 5,
            "error_count": 0,
            "source_exhausted": true,
            "sampling_applied": false
        }"#;
        let meta: ExecutionMetadata = serde_json::from_str(json).unwrap();
        assert_eq!(meta.ragged_row_count, 0);
        assert_eq!(meta.sampled_row_ranges, None);
        assert_eq!(meta.recovery_events, None);
        assert!(
            serde_json::to_value(&meta)
                .unwrap()
                .get("recovery_events")
                .is_none()
        );
        assert!(
            serde_json::to_value(&meta)
                .unwrap()
                .get("sampled_row_ranges")
                .is_none()
        );
    }

    #[test]
    fn recorded_row_selection_preserves_empty_and_nonempty_ranges() {
        for ranges in [vec![], vec![[0, 1], [99, 100]]] {
            let mut meta = ExecutionMetadata::new(0, 1, 0);
            meta.sampled_row_ranges = Some(ranges.clone());
            let json = serde_json::to_value(&meta).unwrap();
            let restored: ExecutionMetadata = serde_json::from_value(json).unwrap();
            assert_eq!(restored.sampled_row_ranges, Some(ranges));
        }
    }

    #[test]
    fn recovery_history_preserves_absence_empty_and_ordered_events() {
        let events = vec![
            RecoveryEvent {
                kind: RecoveryKind::EngineFallback,
                attempted: "columnar".into(),
                retry: "incremental".into(),
                error: "primary failed".into(),
            },
            RecoveryEvent {
                kind: RecoveryKind::CsvAutoRecovery,
                attempted: "strict".into(),
                retry: "flexible".into(),
                error: "unequal field count".into(),
            },
        ];
        assert_eq!(
            ExecutionMetadata::new(0, 0, 0).recovery_events,
            Some(vec![])
        );
        for history in [None, Some(vec![]), Some(events)] {
            let mut meta = ExecutionMetadata::new(0, 0, 0);
            meta.recovery_events = history.clone();
            let json = serde_json::to_value(&meta).unwrap();
            if let Some(events) = &history {
                for (index, event) in events.iter().enumerate() {
                    assert_eq!(json["recovery_events"][index]["kind"], event.kind.as_str());
                }
            }
            let restored: ExecutionMetadata = serde_json::from_value(json).unwrap();
            assert_eq!(restored.recovery_events, history);
        }
    }

    #[test]
    fn test_execution_metadata_with_truncation() {
        let meta =
            ExecutionMetadata::new(1000, 5, 200).with_truncation(TruncationReason::MaxRows(1000));
        assert!(!meta.source_exhausted);
        assert!(meta.truncation_reason.is_some());
    }

    #[test]
    fn test_truncation_reason_serde_roundtrip() {
        let reasons = vec![
            TruncationReason::MaxRows(5000),
            TruncationReason::MaxBytes(1_000_000),
            TruncationReason::MemoryPressure,
            TruncationReason::StopCondition("accuracy > 0.95".to_string()),
            TruncationReason::StreamClosed,
            TruncationReason::Timeout,
        ];

        for reason in reasons {
            let json = serde_json::to_string(&reason).unwrap();
            let deserialized: TruncationReason = serde_json::from_str(&json).unwrap();
            let json2 = serde_json::to_string(&deserialized).unwrap();
            assert_eq!(json, json2);
        }
    }
}