delta_funnel/report/delta/
source.rs

1use std::fmt;
2
3use crate::{
4    DeltaProtocolReport, DeltaProviderReadStatsSnapshot, DeltaProviderReaderBackend,
5    DeltaProviderScanExecutionOptions, FileCount, PhaseTimingReport, QueryOptions,
6    ReportReasonCode, support::sanitize_uri_for_display,
7};
8
9/// Conservative source usage status for a workflow report.
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11pub enum SourceUsageStatus {
12    /// Source usage is known and the source was used by at least one selected output.
13    Used,
14    /// Source usage is known and the source was not used by selected outputs.
15    NotUsed,
16    /// Source usage could not be proven from available workflow analysis.
17    Unknown,
18}
19
20impl SourceUsageStatus {
21    /// Returns a stable lower-snake-case code for report serialization.
22    #[must_use]
23    pub const fn as_str(self) -> &'static str {
24        match self {
25            Self::Used => "used",
26            Self::NotUsed => "not_used",
27            Self::Unknown => "unknown",
28        }
29    }
30}
31
32impl fmt::Display for SourceUsageStatus {
33    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
34        formatter.write_str(self.as_str())
35    }
36}
37
38/// Source-level readiness report for one registered Delta source.
39#[derive(Debug, Clone, PartialEq, Eq)]
40pub struct DeltaSourceReport {
41    source_name: String,
42    source_uri: String,
43    snapshot_version: u64,
44    protocol: DeltaProtocolReport,
45    scheduling: DeltaProviderSchedulingReport,
46    file_count: FileCount,
47    file_count_reason: Option<ReportReasonCode>,
48    scan_metadata_exhausted: bool,
49    usage_status: SourceUsageStatus,
50    used_by_output_names: Vec<String>,
51    provider_read_stats: Option<DeltaProviderReadStatsSnapshot>,
52    provider_stats_reason: Option<ReportReasonCode>,
53    phase_timings: Vec<PhaseTimingReport>,
54}
55
56/// Configured provider scheduling options included with a Delta source report.
57#[derive(Debug, Clone, Copy, PartialEq, Eq)]
58pub struct DeltaProviderSchedulingReport {
59    query_target_partitions: Option<u64>,
60    reader_backend: DeltaProviderReaderBackend,
61    max_concurrent_file_reads_per_scan: Option<u64>,
62    max_concurrent_file_reads_per_partition: u64,
63    output_buffer_capacity_per_partition: u64,
64    native_async_prefetch_file_count_per_partition: u64,
65}
66
67impl DeltaProviderSchedulingReport {
68    pub(crate) fn from_options(
69        query_options: QueryOptions,
70        scan_options: DeltaProviderScanExecutionOptions,
71    ) -> Self {
72        Self {
73            query_target_partitions: query_options
74                .target_partitions
75                .map(crate::usize_to_u64_saturating),
76            reader_backend: scan_options.reader_backend,
77            max_concurrent_file_reads_per_scan: scan_options
78                .max_concurrent_file_reads_per_scan
79                .map(crate::usize_to_u64_saturating),
80            max_concurrent_file_reads_per_partition: crate::usize_to_u64_saturating(
81                scan_options.max_concurrent_file_reads_per_partition,
82            ),
83            output_buffer_capacity_per_partition: crate::usize_to_u64_saturating(
84                scan_options.output_buffer_capacity_per_partition,
85            ),
86            native_async_prefetch_file_count_per_partition: crate::usize_to_u64_saturating(
87                scan_options.native_async_prefetch_file_count_per_partition,
88            ),
89        }
90    }
91
92    /// Returns the configured DataFusion target partition count, when set.
93    #[must_use]
94    pub const fn query_target_partitions(&self) -> Option<u64> {
95        self.query_target_partitions
96    }
97
98    /// Returns the provider reader backend configured for source scans.
99    #[must_use]
100    pub const fn reader_backend(&self) -> DeltaProviderReaderBackend {
101        self.reader_backend
102    }
103
104    /// Returns the configured scan-wide file-read concurrency cap.
105    ///
106    /// `None` means the cap is resolved during physical scan planning from the
107    /// derived scan partition target and the configured per-partition cap.
108    #[must_use]
109    pub const fn max_concurrent_file_reads_per_scan(&self) -> Option<u64> {
110        self.max_concurrent_file_reads_per_scan
111    }
112
113    /// Returns the configured per-partition file-read concurrency cap.
114    #[must_use]
115    pub const fn max_concurrent_file_reads_per_partition(&self) -> u64 {
116        self.max_concurrent_file_reads_per_partition
117    }
118
119    /// Returns the configured per-partition provider output buffer capacity.
120    #[must_use]
121    pub const fn output_buffer_capacity_per_partition(&self) -> u64 {
122        self.output_buffer_capacity_per_partition
123    }
124
125    /// Returns the native async prefetch depth per execution partition.
126    #[must_use]
127    pub const fn native_async_prefetch_file_count_per_partition(&self) -> u64 {
128        self.native_async_prefetch_file_count_per_partition
129    }
130}
131
132impl DeltaSourceReport {
133    pub(crate) fn metadata_only(
134        source_name: impl Into<String>,
135        source_uri: impl Into<String>,
136        snapshot_version: u64,
137        protocol: DeltaProtocolReport,
138        scheduling: DeltaProviderSchedulingReport,
139    ) -> Self {
140        Self {
141            source_name: source_name.into(),
142            source_uri: sanitize_uri_for_display(&source_uri.into()),
143            snapshot_version,
144            protocol,
145            scheduling,
146            file_count: FileCount::unavailable(),
147            file_count_reason: Some(ReportReasonCode::CostAvoidance),
148            scan_metadata_exhausted: false,
149            usage_status: SourceUsageStatus::Unknown,
150            used_by_output_names: Vec::new(),
151            provider_read_stats: None,
152            provider_stats_reason: Some(ReportReasonCode::NotExecuted),
153            phase_timings: Vec::new(),
154        }
155    }
156
157    pub(crate) fn with_phase_timings(mut self, phase_timings: Vec<PhaseTimingReport>) -> Self {
158        self.phase_timings = phase_timings;
159        self
160    }
161
162    pub(crate) fn with_usage(
163        mut self,
164        usage_status: SourceUsageStatus,
165        used_by_output_names: Vec<String>,
166    ) -> Self {
167        self.usage_status = usage_status;
168        self.used_by_output_names = used_by_output_names;
169        self
170    }
171
172    pub(crate) fn with_provider_read_stats(
173        mut self,
174        stats: DeltaProviderReadStatsSnapshot,
175    ) -> Self {
176        self.scan_metadata_exhausted = stats.scan_metadata_exhausted.unwrap_or(false);
177        self.file_count = match stats.scan_metadata_exhausted {
178            Some(true) => FileCount::exact(stats.files_planned),
179            Some(false) => FileCount::estimated(stats.files_planned),
180            None => FileCount::unavailable(),
181        };
182        self.file_count_reason = match self.file_count {
183            FileCount::Exact(_) | FileCount::Estimated(_) => None,
184            FileCount::Unavailable => Some(ReportReasonCode::CapabilityUnavailable),
185            FileCount::Skipped | FileCount::NotExecuted => Some(ReportReasonCode::NotExecuted),
186        };
187        self.provider_read_stats = Some(stats);
188        self.provider_stats_reason = None;
189        self
190    }
191
192    pub(crate) fn with_provider_stats_reason(mut self, reason: ReportReasonCode) -> Self {
193        self.provider_read_stats = None;
194        self.provider_stats_reason = Some(reason);
195        self
196    }
197
198    /// Returns the DataFusion table name for this source.
199    #[must_use]
200    pub fn source_name(&self) -> &str {
201        &self.source_name
202    }
203
204    /// Returns the sanitized Delta source URI or display summary.
205    #[must_use]
206    pub fn source_uri(&self) -> &str {
207        &self.source_uri
208    }
209
210    /// Returns the resolved Delta snapshot version.
211    #[must_use]
212    pub const fn snapshot_version(&self) -> u64 {
213        self.snapshot_version
214    }
215
216    /// Returns the Delta protocol report captured for this source.
217    #[must_use]
218    pub const fn protocol(&self) -> &DeltaProtocolReport {
219        &self.protocol
220    }
221
222    /// Returns the configured provider scheduling options for this source.
223    #[must_use]
224    pub const fn scheduling(&self) -> &DeltaProviderSchedulingReport {
225        &self.scheduling
226    }
227
228    /// Returns file-count evidence for this source.
229    #[must_use]
230    pub const fn file_count(&self) -> FileCount {
231        self.file_count
232    }
233
234    /// Returns the stable reason code for unavailable, skipped, or not-executed file counts.
235    #[must_use]
236    pub const fn file_count_reason(&self) -> Option<ReportReasonCode> {
237        self.file_count_reason
238    }
239
240    /// Returns whether scan metadata was exhausted while building this report.
241    #[must_use]
242    pub const fn scan_metadata_exhausted(&self) -> bool {
243        self.scan_metadata_exhausted
244    }
245
246    /// Returns known source usage status for the workflow scope that produced this report.
247    #[must_use]
248    pub const fn usage_status(&self) -> SourceUsageStatus {
249        self.usage_status
250    }
251
252    /// Returns selected output names known to use this source.
253    #[must_use]
254    pub fn used_by_output_names(&self) -> &[String] {
255        &self.used_by_output_names
256    }
257
258    /// Returns provider read statistics when planning or execution made them available.
259    #[must_use]
260    pub const fn provider_read_stats(&self) -> Option<&DeltaProviderReadStatsSnapshot> {
261        self.provider_read_stats.as_ref()
262    }
263
264    /// Returns the stable reason code when provider read statistics are absent.
265    #[must_use]
266    pub const fn provider_stats_reason(&self) -> Option<ReportReasonCode> {
267        self.provider_stats_reason
268    }
269
270    /// Returns durable phase timings captured while registering this source.
271    #[must_use]
272    pub fn phase_timings(&self) -> &[PhaseTimingReport] {
273        &self.phase_timings
274    }
275}
276
277#[cfg(test)]
278mod tests {
279    use super::SourceUsageStatus;
280
281    #[test]
282    fn source_usage_status_exposes_stable_codes() {
283        assert_eq!(SourceUsageStatus::Used.as_str(), "used");
284        assert_eq!(SourceUsageStatus::NotUsed.as_str(), "not_used");
285        assert_eq!(SourceUsageStatus::Unknown.as_str(), "unknown");
286        assert_eq!(SourceUsageStatus::Unknown.to_string(), "unknown");
287    }
288}
delta_funnel/report/delta/source.rs

delta_funnel/report/delta/
source.rs
