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
use std::fmt;
use std::sync::Arc;
use std::time::Duration;
/// Structured progress events emitted by profiling engines.
#[derive(Debug, Clone)]
pub enum ProgressEvent {
/// Profiling has started. Emitted once before any data is processed.
Started {
estimated_total_rows: Option<usize>,
estimated_total_bytes: Option<u64>,
},
/// A chunk of rows has been processed.
ChunkProcessed {
rows_processed: usize,
bytes_consumed: u64,
elapsed: Duration,
processing_speed: f64,
percentage: Option<f64>,
},
/// Column schema has been detected: emitted after the first chunk, and
/// again with the full list whenever a later column joins (an async JSON
/// key first seen after the first chunk). Each list extends the last.
SchemaDetected { column_names: Vec<String> },
/// Profiling has finished (successfully or via early stop).
Finished {
total_rows: usize,
total_bytes: u64,
elapsed: Duration,
truncated: bool,
},
/// A non-fatal warning occurred during processing.
Warning { message: String },
}
/// How progress events are delivered to the consumer.
///
/// All variants are cheaply cloneable (`Arc` / channel `Sender`).
#[derive(Clone, Default)]
pub enum ProgressSink {
/// No progress reporting.
#[default]
None,
/// Synchronous callback (for Python bindings, tests, embedding, etc.)
Callback(Arc<dyn Fn(ProgressEvent) + Send + Sync>),
/// Async channel sender (requires `async-streaming` feature).
#[cfg(feature = "async-streaming")]
Channel(tokio::sync::mpsc::Sender<ProgressEvent>),
}
impl ProgressSink {
/// Whether events go anywhere.
pub fn is_active(&self) -> bool {
!matches!(self, Self::None)
}
/// Reserve channel capacity for a terminal event without changing the
/// public shape of the Channel variant.
#[doc(hidden)]
#[cfg(feature = "async-streaming")]
pub fn reserve_terminal(&self) -> Option<tokio::sync::mpsc::OwnedPermit<ProgressEvent>> {
match self {
Self::Channel(tx) => tx.clone().try_reserve_owned().ok(),
_ => None,
}
}
/// Deliver one event. A full channel drops it rather than block profiling.
pub fn emit(&self, event: ProgressEvent) {
match self {
Self::None => {}
Self::Callback(callback) => callback(event),
#[cfg(feature = "async-streaming")]
Self::Channel(tx) => {
let _ = tx.try_send(event);
}
}
}
}
impl fmt::Debug for ProgressSink {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::None => write!(f, "ProgressSink::None"),
Self::Callback(_) => write!(f, "ProgressSink::Callback(..)"),
#[cfg(feature = "async-streaming")]
Self::Channel(_) => write!(f, "ProgressSink::Channel(..)"),
}
}
}