dataprof_core/
progress.rs

1use std::fmt;
2use std::sync::Arc;
3use std::time::Duration;
4
5/// Structured progress events emitted by profiling engines.
6#[derive(Debug, Clone)]
7pub enum ProgressEvent {
8    /// Profiling has started. Emitted once before any data is processed.
9    Started {
10        estimated_total_rows: Option<usize>,
11        estimated_total_bytes: Option<u64>,
12    },
13    /// A chunk of rows has been processed.
14    ChunkProcessed {
15        rows_processed: usize,
16        bytes_consumed: u64,
17        elapsed: Duration,
18        processing_speed: f64,
19        percentage: Option<f64>,
20    },
21    /// Column schema has been detected: emitted after the first chunk, and
22    /// again with the full list whenever a later column joins (an async JSON
23    /// key first seen after the first chunk). Each list extends the last.
24    SchemaDetected { column_names: Vec<String> },
25    /// Profiling has finished (successfully or via early stop).
26    Finished {
27        total_rows: usize,
28        total_bytes: u64,
29        elapsed: Duration,
30        truncated: bool,
31    },
32    /// A non-fatal warning occurred during processing.
33    Warning { message: String },
34}
35
36/// How progress events are delivered to the consumer.
37///
38/// All variants are cheaply cloneable (`Arc` / channel `Sender`).
39#[derive(Clone, Default)]
40pub enum ProgressSink {
41    /// No progress reporting.
42    #[default]
43    None,
44    /// Synchronous callback (for Python bindings, tests, embedding, etc.)
45    Callback(Arc<dyn Fn(ProgressEvent) + Send + Sync>),
46    /// Async channel sender (requires `async-streaming` feature).
47    #[cfg(feature = "async-streaming")]
48    Channel(tokio::sync::mpsc::Sender<ProgressEvent>),
49}
50
51impl ProgressSink {
52    /// Whether events go anywhere.
53    pub fn is_active(&self) -> bool {
54        !matches!(self, Self::None)
55    }
56
57    /// Reserve channel capacity for a terminal event without changing the
58    /// public shape of the Channel variant.
59    #[doc(hidden)]
60    #[cfg(feature = "async-streaming")]
61    pub fn reserve_terminal(&self) -> Option<tokio::sync::mpsc::OwnedPermit<ProgressEvent>> {
62        match self {
63            Self::Channel(tx) => tx.clone().try_reserve_owned().ok(),
64            _ => None,
65        }
66    }
67
68    /// Deliver one event. A full channel drops it rather than block profiling.
69    pub fn emit(&self, event: ProgressEvent) {
70        match self {
71            Self::None => {}
72            Self::Callback(callback) => callback(event),
73            #[cfg(feature = "async-streaming")]
74            Self::Channel(tx) => {
75                let _ = tx.try_send(event);
76            }
77        }
78    }
79}
80
81impl fmt::Debug for ProgressSink {
82    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
83        match self {
84            Self::None => write!(f, "ProgressSink::None"),
85            Self::Callback(_) => write!(f, "ProgressSink::Callback(..)"),
86            #[cfg(feature = "async-streaming")]
87            Self::Channel(_) => write!(f, "ProgressSink::Channel(..)"),
88        }
89    }
90}
dataprof_core/progress.rs

dataprof_core/
progress.rs
