Skip to main content

delta_arrow_reader/reader/
metrics.rs

1//! Delta scan planning and execution metrics.
2
3use std::{
4    fmt,
5    sync::{
6        Arc,
7        atomic::{AtomicU64, Ordering},
8    },
9};
10
11use super::options::MAX_CONCURRENT_PARQUET_RANGE_READS;
12use super::options::ParquetReaderBackend;
13
14/// Internal measurements used to validate Parquet range-planning benchmarks.
15#[doc(hidden)]
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub struct ParquetRangePlanningDiagnosticSnapshot {
18    /// Largest per-plan concurrency allowance, starting at the ordinary range-read limit.
19    pub max_concurrent_physical_range_requests: u64,
20    /// Sum of physical request waves selected by visible plans.
21    pub physical_range_request_waves_planned: u64,
22    /// Sum of successful multi-range plan execution time in microseconds.
23    pub successful_plan_time_micros: u64,
24}
25
26/// Immutable point-in-time copy of one Delta scan's metrics.
27///
28/// A snapshot taken while the scan is running may contain partial progress. It does not change
29/// after creation; call [`DeltaScanMetrics::snapshot`] again to observe later progress.
30#[non_exhaustive]
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct DeltaScanMetricsSnapshot {
33    /// Delta snapshot version selected for the scan.
34    pub snapshot_version: u64,
35    /// Parquet reader backend selected for the scan.
36    pub parquet_backend: ParquetReaderBackend,
37    /// Final execution partitions planned for the scan, including source repartitioning.
38    pub scan_partitions_planned: u64,
39    /// Data files selected during planning.
40    pub files_planned: u64,
41    /// Add actions excluded during planning, when known.
42    pub add_actions_excluded_during_planning: Option<u64>,
43    /// Estimated rows in selected input files before row filtering, when known for every file.
44    pub estimated_input_rows: Option<u64>,
45    /// Estimated bytes in selected input files, when every file reported a size.
46    pub estimated_input_bytes: Option<u64>,
47    /// Scan partitions whose execution started.
48    pub scan_partitions_started: u64,
49    /// Scan partitions that completed normally.
50    pub scan_partitions_completed: u64,
51    /// Data-file tasks that started, including independently read file ranges.
52    pub file_tasks_started: u64,
53    /// Data-file tasks that completed normally, including independently read file ranges.
54    pub file_tasks_completed: u64,
55    /// Batches emitted by the scheduler before final stream operations.
56    pub scheduler_batches_emitted: u64,
57    /// Rows emitted by the scheduler before final stream operations.
58    pub scheduler_rows_emitted: u64,
59    /// Deletion-vector payloads loaded.
60    pub deletion_vector_payloads_loaded: u64,
61    /// Deletion-vector masks applied.
62    pub deletion_vectors_applied: u64,
63    /// Rows removed by deletion-vector masks.
64    pub deletion_vector_rows_deleted: u64,
65    /// Deletion-vector read or masking failures.
66    pub deletion_vector_failures: u64,
67    /// Deletion-vector coordinate operations rejected by safety checks.
68    pub deletion_vector_coordinate_rejections: u64,
69    /// Normalized exact Parquet ranges requested through direct multi-range calls.
70    pub parquet_data_file_exact_ranges_requested: Option<u64>,
71    /// Bytes in those normalized exact Parquet ranges.
72    pub parquet_data_file_exact_range_bytes_requested: Option<u64>,
73    /// Physical requests selected by visible range plans, including partial-page reads.
74    pub parquet_data_file_physical_range_requests_planned: Option<u64>,
75    /// Bytes covered by those planned physical range requests.
76    pub parquet_data_file_physical_range_bytes_planned: Option<u64>,
77    /// Automatic range plans selected without a usable transport estimate.
78    pub parquet_data_file_cold_start_range_plans: Option<u64>,
79    /// Automatic range plans where an estimate favored the normalized exact ranges.
80    pub parquet_data_file_cost_based_exact_range_plans: Option<u64>,
81    /// Automatic range plans where an estimate favored merging gaps between exact ranges.
82    pub parquet_data_file_cost_based_merged_range_plans: Option<u64>,
83    /// Range-planning decisions delegated to the store's own implementation.
84    pub parquet_data_file_store_delegated_range_plans: Option<u64>,
85    /// Direct Parquet ranged GET operations, or `None` for another backend.
86    pub parquet_data_file_range_get_operations: Option<u64>,
87    /// Direct Parquet full GET operations, or `None` for another backend.
88    pub parquet_data_file_full_get_operations: Option<u64>,
89    /// Direct Parquet payload bytes received, or `None` for another backend.
90    pub parquet_data_file_bytes_received: Option<u64>,
91    /// Estimated bytes admitted across direct Parquet tasks, or `None` for another backend.
92    pub estimated_parquet_task_bytes_admitted: Option<u64>,
93}
94
95/// Shared live metrics for one Delta scan.
96#[derive(Clone)]
97pub struct DeltaScanMetrics {
98    inner: Arc<DeltaScanMetricsInner>,
99}
100
101impl fmt::Debug for DeltaScanMetrics {
102    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
103        formatter
104            .debug_struct("DeltaScanMetrics")
105            .finish_non_exhaustive()
106    }
107}
108
109struct DeltaScanMetricsInner {
110    snapshot_version: u64,
111    parquet_backend: ParquetReaderBackend,
112    scan_partitions_planned: AtomicU64,
113    files_planned: u64,
114    add_actions_excluded_during_planning: Option<u64>,
115    estimated_input_rows: Option<u64>,
116    estimated_input_bytes: Option<u64>,
117    scan_partitions_started: AtomicU64,
118    scan_partitions_completed: AtomicU64,
119    file_tasks_started: AtomicU64,
120    file_tasks_completed: AtomicU64,
121    scheduler_batches_emitted: AtomicU64,
122    scheduler_rows_emitted: AtomicU64,
123    deletion_vector_payloads_loaded: AtomicU64,
124    deletion_vectors_applied: AtomicU64,
125    deletion_vector_rows_deleted: AtomicU64,
126    deletion_vector_failures: AtomicU64,
127    deletion_vector_coordinate_rejections: AtomicU64,
128    parquet_data_file_exact_ranges_requested: AtomicU64,
129    parquet_data_file_exact_range_bytes_requested: AtomicU64,
130    parquet_data_file_physical_range_requests_planned: AtomicU64,
131    parquet_data_file_physical_range_bytes_planned: AtomicU64,
132    parquet_data_file_cold_start_range_plans: AtomicU64,
133    parquet_data_file_cost_based_exact_range_plans: AtomicU64,
134    parquet_data_file_cost_based_merged_range_plans: AtomicU64,
135    parquet_data_file_store_delegated_range_plans: AtomicU64,
136    parquet_range_request_waves_planned: AtomicU64,
137    parquet_range_concurrency_limit: AtomicU64,
138    parquet_range_successful_plan_time_micros: AtomicU64,
139    parquet_data_file_range_get_operations: AtomicU64,
140    parquet_data_file_full_get_operations: AtomicU64,
141    parquet_data_file_bytes_received: AtomicU64,
142    estimated_parquet_task_bytes_admitted: AtomicU64,
143}
144
145#[allow(dead_code)]
146pub(crate) struct DeltaScanMetricsConfig {
147    pub(crate) snapshot_version: u64,
148    pub(crate) parquet_backend: ParquetReaderBackend,
149    pub(crate) scan_partitions_planned: usize,
150    pub(crate) files_planned: usize,
151    pub(crate) add_actions_excluded_during_planning: Option<u64>,
152    pub(crate) estimated_input_rows: Option<u64>,
153    pub(crate) estimated_input_bytes: Option<u64>,
154}
155
156impl DeltaScanMetrics {
157    #[allow(dead_code)]
158    pub(crate) fn new(config: DeltaScanMetricsConfig) -> Self {
159        Self {
160            inner: Arc::new(DeltaScanMetricsInner {
161                snapshot_version: config.snapshot_version,
162                parquet_backend: config.parquet_backend,
163                scan_partitions_planned: AtomicU64::new(usize_to_u64_saturating(
164                    config.scan_partitions_planned,
165                )),
166                files_planned: usize_to_u64_saturating(config.files_planned),
167                add_actions_excluded_during_planning: config.add_actions_excluded_during_planning,
168                estimated_input_rows: config.estimated_input_rows,
169                estimated_input_bytes: config.estimated_input_bytes,
170                scan_partitions_started: AtomicU64::new(0),
171                scan_partitions_completed: AtomicU64::new(0),
172                file_tasks_started: AtomicU64::new(0),
173                file_tasks_completed: AtomicU64::new(0),
174                scheduler_batches_emitted: AtomicU64::new(0),
175                scheduler_rows_emitted: AtomicU64::new(0),
176                deletion_vector_payloads_loaded: AtomicU64::new(0),
177                deletion_vectors_applied: AtomicU64::new(0),
178                deletion_vector_rows_deleted: AtomicU64::new(0),
179                deletion_vector_failures: AtomicU64::new(0),
180                deletion_vector_coordinate_rejections: AtomicU64::new(0),
181                parquet_data_file_exact_ranges_requested: AtomicU64::new(0),
182                parquet_data_file_exact_range_bytes_requested: AtomicU64::new(0),
183                parquet_data_file_physical_range_requests_planned: AtomicU64::new(0),
184                parquet_data_file_physical_range_bytes_planned: AtomicU64::new(0),
185                parquet_data_file_cold_start_range_plans: AtomicU64::new(0),
186                parquet_data_file_cost_based_exact_range_plans: AtomicU64::new(0),
187                parquet_data_file_cost_based_merged_range_plans: AtomicU64::new(0),
188                parquet_data_file_store_delegated_range_plans: AtomicU64::new(0),
189                parquet_range_request_waves_planned: AtomicU64::new(0),
190                parquet_range_concurrency_limit: AtomicU64::new(usize_to_u64_saturating(
191                    MAX_CONCURRENT_PARQUET_RANGE_READS,
192                )),
193                parquet_range_successful_plan_time_micros: AtomicU64::new(0),
194                parquet_data_file_range_get_operations: AtomicU64::new(0),
195                parquet_data_file_full_get_operations: AtomicU64::new(0),
196                parquet_data_file_bytes_received: AtomicU64::new(0),
197                estimated_parquet_task_bytes_admitted: AtomicU64::new(0),
198            }),
199        }
200    }
201
202    /// Returns an immutable point-in-time copy of all scan metrics.
203    pub fn snapshot(&self) -> DeltaScanMetricsSnapshot {
204        let inner = self.inner.as_ref();
205        DeltaScanMetricsSnapshot {
206            snapshot_version: inner.snapshot_version,
207            parquet_backend: inner.parquet_backend,
208            scan_partitions_planned: load(&inner.scan_partitions_planned),
209            files_planned: inner.files_planned,
210            add_actions_excluded_during_planning: inner.add_actions_excluded_during_planning,
211            estimated_input_rows: inner.estimated_input_rows,
212            estimated_input_bytes: inner.estimated_input_bytes,
213            scan_partitions_started: load(&inner.scan_partitions_started),
214            scan_partitions_completed: load(&inner.scan_partitions_completed),
215            file_tasks_started: load(&inner.file_tasks_started),
216            file_tasks_completed: load(&inner.file_tasks_completed),
217            scheduler_batches_emitted: load(&inner.scheduler_batches_emitted),
218            scheduler_rows_emitted: load(&inner.scheduler_rows_emitted),
219            deletion_vector_payloads_loaded: load(&inner.deletion_vector_payloads_loaded),
220            deletion_vectors_applied: load(&inner.deletion_vectors_applied),
221            deletion_vector_rows_deleted: load(&inner.deletion_vector_rows_deleted),
222            deletion_vector_failures: load(&inner.deletion_vector_failures),
223            deletion_vector_coordinate_rejections: load(
224                &inner.deletion_vector_coordinate_rejections,
225            ),
226            parquet_data_file_exact_ranges_requested: self
227                .parquet_metric(&inner.parquet_data_file_exact_ranges_requested),
228            parquet_data_file_exact_range_bytes_requested: self
229                .parquet_metric(&inner.parquet_data_file_exact_range_bytes_requested),
230            parquet_data_file_physical_range_requests_planned: self
231                .parquet_metric(&inner.parquet_data_file_physical_range_requests_planned),
232            parquet_data_file_physical_range_bytes_planned: self
233                .parquet_metric(&inner.parquet_data_file_physical_range_bytes_planned),
234            parquet_data_file_cold_start_range_plans: self
235                .parquet_metric(&inner.parquet_data_file_cold_start_range_plans),
236            parquet_data_file_cost_based_exact_range_plans: self
237                .parquet_metric(&inner.parquet_data_file_cost_based_exact_range_plans),
238            parquet_data_file_cost_based_merged_range_plans: self
239                .parquet_metric(&inner.parquet_data_file_cost_based_merged_range_plans),
240            parquet_data_file_store_delegated_range_plans: self
241                .parquet_metric(&inner.parquet_data_file_store_delegated_range_plans),
242            parquet_data_file_range_get_operations: self
243                .parquet_metric(&inner.parquet_data_file_range_get_operations),
244            parquet_data_file_full_get_operations: self
245                .parquet_metric(&inner.parquet_data_file_full_get_operations),
246            parquet_data_file_bytes_received: self
247                .parquet_metric(&inner.parquet_data_file_bytes_received),
248            estimated_parquet_task_bytes_admitted: self
249                .parquet_metric(&inner.estimated_parquet_task_bytes_admitted),
250        }
251    }
252
253    fn parquet_metric(&self, counter: &AtomicU64) -> Option<u64> {
254        match self.inner.parquet_backend {
255            ParquetReaderBackend::Direct => Some(load(counter)),
256            ParquetReaderBackend::DeltaKernel => None,
257        }
258    }
259
260    pub(crate) fn parquet_range_planning_diagnostic_snapshot(
261        &self,
262    ) -> ParquetRangePlanningDiagnosticSnapshot {
263        ParquetRangePlanningDiagnosticSnapshot {
264            max_concurrent_physical_range_requests: load(
265                &self.inner.parquet_range_concurrency_limit,
266            ),
267            physical_range_request_waves_planned: load(
268                &self.inner.parquet_range_request_waves_planned,
269            ),
270            successful_plan_time_micros: load(
271                &self.inner.parquet_range_successful_plan_time_micros,
272            ),
273        }
274    }
275
276    #[allow(dead_code)]
277    pub(crate) fn record_scan_partitions_planned(&self, value: usize) {
278        self.inner
279            .scan_partitions_planned
280            .store(usize_to_u64_saturating(value), Ordering::Relaxed);
281    }
282
283    #[allow(dead_code)]
284    pub(crate) fn record_scan_partition_started(&self) {
285        saturating_fetch_add(&self.inner.scan_partitions_started, 1);
286    }
287
288    #[allow(dead_code)]
289    pub(crate) fn record_scan_partition_completed(&self) {
290        saturating_fetch_add(&self.inner.scan_partitions_completed, 1);
291    }
292
293    #[allow(dead_code)]
294    pub(crate) fn record_file_task_started(&self) {
295        saturating_fetch_add(&self.inner.file_tasks_started, 1);
296    }
297
298    #[allow(dead_code)]
299    pub(crate) fn record_file_task_completed(&self) {
300        saturating_fetch_add(&self.inner.file_tasks_completed, 1);
301    }
302
303    #[allow(dead_code)]
304    pub(crate) fn record_scheduler_batch_emitted(&self, rows: usize) {
305        saturating_fetch_add(&self.inner.scheduler_batches_emitted, 1);
306        saturating_fetch_add(
307            &self.inner.scheduler_rows_emitted,
308            usize_to_u64_saturating(rows),
309        );
310    }
311
312    #[allow(dead_code)]
313    pub(crate) fn record_deletion_vector_payload_loaded(&self) {
314        saturating_fetch_add(&self.inner.deletion_vector_payloads_loaded, 1);
315    }
316
317    #[allow(dead_code)]
318    pub(crate) fn record_deletion_vector_applied(&self) {
319        saturating_fetch_add(&self.inner.deletion_vectors_applied, 1);
320    }
321
322    #[allow(dead_code)]
323    pub(crate) fn record_deletion_vector_rows_deleted(&self, rows: usize) {
324        saturating_fetch_add(
325            &self.inner.deletion_vector_rows_deleted,
326            usize_to_u64_saturating(rows),
327        );
328    }
329
330    #[allow(dead_code)]
331    pub(crate) fn record_deletion_vector_failure(&self) {
332        saturating_fetch_add(&self.inner.deletion_vector_failures, 1);
333    }
334
335    #[allow(dead_code)]
336    pub(crate) fn record_deletion_vector_coordinate_rejection(&self) {
337        saturating_fetch_add(&self.inner.deletion_vector_coordinate_rejections, 1);
338    }
339
340    pub(crate) fn record_parquet_data_file_exact_ranges_requested(
341        &self,
342        range_count: usize,
343        bytes: u128,
344    ) {
345        saturating_fetch_add(
346            &self.inner.parquet_data_file_exact_ranges_requested,
347            usize_to_u64_saturating(range_count),
348        );
349        saturating_fetch_add(
350            &self.inner.parquet_data_file_exact_range_bytes_requested,
351            u128_to_u64_saturating(bytes),
352        );
353    }
354
355    pub(crate) fn record_parquet_data_file_physical_range_plan(
356        &self,
357        request_count: usize,
358        bytes: u128,
359        concurrency: usize,
360    ) {
361        saturating_fetch_add(
362            &self.inner.parquet_data_file_physical_range_requests_planned,
363            usize_to_u64_saturating(request_count),
364        );
365        saturating_fetch_add(
366            &self.inner.parquet_data_file_physical_range_bytes_planned,
367            u128_to_u64_saturating(bytes),
368        );
369        saturating_fetch_add(
370            &self.inner.parquet_range_request_waves_planned,
371            usize_to_u64_saturating(request_count.div_ceil(concurrency.max(1))),
372        );
373        self.inner
374            .parquet_range_concurrency_limit
375            .fetch_max(usize_to_u64_saturating(concurrency), Ordering::Relaxed);
376    }
377
378    pub(crate) fn record_parquet_range_successful_plan_time(&self, elapsed: std::time::Duration) {
379        saturating_fetch_add(
380            &self.inner.parquet_range_successful_plan_time_micros,
381            u128_to_u64_saturating(elapsed.as_micros()),
382        );
383    }
384
385    pub(crate) fn record_parquet_data_file_cold_start_range_plan(&self) {
386        saturating_fetch_add(&self.inner.parquet_data_file_cold_start_range_plans, 1);
387    }
388
389    pub(crate) fn record_parquet_data_file_cost_based_exact_range_plan(&self) {
390        saturating_fetch_add(
391            &self.inner.parquet_data_file_cost_based_exact_range_plans,
392            1,
393        );
394    }
395
396    pub(crate) fn record_parquet_data_file_cost_based_merged_range_plan(&self) {
397        saturating_fetch_add(
398            &self.inner.parquet_data_file_cost_based_merged_range_plans,
399            1,
400        );
401    }
402
403    pub(crate) fn record_parquet_data_file_store_delegated_range_plan(&self) {
404        saturating_fetch_add(&self.inner.parquet_data_file_store_delegated_range_plans, 1);
405    }
406
407    pub(crate) fn record_parquet_data_file_range_get_operation(&self) {
408        saturating_fetch_add(&self.inner.parquet_data_file_range_get_operations, 1);
409    }
410
411    pub(crate) fn record_parquet_data_file_full_get_operation(&self) {
412        saturating_fetch_add(&self.inner.parquet_data_file_full_get_operations, 1);
413    }
414
415    pub(crate) fn record_parquet_data_file_bytes_received(&self, bytes: usize) {
416        saturating_fetch_add(
417            &self.inner.parquet_data_file_bytes_received,
418            usize_to_u64_saturating(bytes),
419        );
420    }
421
422    pub(crate) fn record_estimated_parquet_task_bytes_admitted(&self, bytes: u64) {
423        saturating_fetch_add(&self.inner.estimated_parquet_task_bytes_admitted, bytes);
424    }
425}
426
427fn load(counter: &AtomicU64) -> u64 {
428    counter.load(Ordering::Relaxed)
429}
430
431#[allow(dead_code)]
432pub(crate) fn saturating_fetch_add(counter: &AtomicU64, value: u64) {
433    let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
434        Some(current.saturating_add(value))
435    });
436}
437
438fn usize_to_u64_saturating(value: usize) -> u64 {
439    u64::try_from(value).unwrap_or(u64::MAX)
440}
441
442fn u128_to_u64_saturating(value: u128) -> u64 {
443    u64::try_from(value).unwrap_or(u64::MAX)
444}
445
446#[cfg(test)]
447mod tests {
448    use std::{sync::atomic::Ordering, thread};
449
450    use super::{DeltaScanMetrics, DeltaScanMetricsConfig, saturating_fetch_add};
451    use crate::ParquetReaderBackend;
452
453    fn metrics(parquet_backend: ParquetReaderBackend) -> DeltaScanMetrics {
454        DeltaScanMetrics::new(DeltaScanMetricsConfig {
455            snapshot_version: 7,
456            parquet_backend,
457            scan_partitions_planned: 3,
458            files_planned: 5,
459            add_actions_excluded_during_planning: Some(2),
460            estimated_input_rows: Some(99),
461            estimated_input_bytes: Some(42),
462        })
463    }
464
465    #[test]
466    fn snapshot_has_context_zeroes_and_backend_availability() {
467        let direct = metrics(ParquetReaderBackend::Direct).snapshot();
468        assert_eq!(direct.snapshot_version, 7);
469        assert_eq!(direct.parquet_backend, ParquetReaderBackend::Direct);
470        assert_eq!(direct.scan_partitions_planned, 3);
471        assert_eq!(direct.files_planned, 5);
472        assert_eq!(direct.add_actions_excluded_during_planning, Some(2));
473        assert_eq!(direct.estimated_input_rows, Some(99));
474        assert_eq!(direct.estimated_input_bytes, Some(42));
475        assert_eq!(direct.scan_partitions_started, 0);
476        assert_eq!(direct.scan_partitions_completed, 0);
477        assert_eq!(direct.file_tasks_started, 0);
478        assert_eq!(direct.file_tasks_completed, 0);
479        assert_eq!(direct.scheduler_batches_emitted, 0);
480        assert_eq!(direct.scheduler_rows_emitted, 0);
481        assert_eq!(direct.deletion_vector_payloads_loaded, 0);
482        assert_eq!(direct.deletion_vectors_applied, 0);
483        assert_eq!(direct.deletion_vector_rows_deleted, 0);
484        assert_eq!(direct.deletion_vector_failures, 0);
485        assert_eq!(direct.deletion_vector_coordinate_rejections, 0);
486        assert_eq!(direct.parquet_data_file_exact_ranges_requested, Some(0));
487        assert_eq!(
488            direct.parquet_data_file_exact_range_bytes_requested,
489            Some(0)
490        );
491        assert_eq!(
492            direct.parquet_data_file_physical_range_requests_planned,
493            Some(0)
494        );
495        assert_eq!(
496            direct.parquet_data_file_physical_range_bytes_planned,
497            Some(0)
498        );
499        assert_eq!(direct.parquet_data_file_cold_start_range_plans, Some(0));
500        assert_eq!(
501            direct.parquet_data_file_cost_based_exact_range_plans,
502            Some(0)
503        );
504        assert_eq!(
505            direct.parquet_data_file_cost_based_merged_range_plans,
506            Some(0)
507        );
508        assert_eq!(
509            direct.parquet_data_file_store_delegated_range_plans,
510            Some(0)
511        );
512        assert_eq!(direct.parquet_data_file_range_get_operations, Some(0));
513        assert_eq!(direct.parquet_data_file_full_get_operations, Some(0));
514        assert_eq!(direct.parquet_data_file_bytes_received, Some(0));
515        assert_eq!(direct.estimated_parquet_task_bytes_admitted, Some(0));
516
517        let kernel = metrics(ParquetReaderBackend::DeltaKernel).snapshot();
518        assert_eq!(kernel.parquet_data_file_exact_ranges_requested, None);
519        assert_eq!(kernel.parquet_data_file_exact_range_bytes_requested, None);
520        assert_eq!(
521            kernel.parquet_data_file_physical_range_requests_planned,
522            None
523        );
524        assert_eq!(kernel.parquet_data_file_physical_range_bytes_planned, None);
525        assert_eq!(kernel.parquet_data_file_cold_start_range_plans, None);
526        assert_eq!(kernel.parquet_data_file_cost_based_exact_range_plans, None);
527        assert_eq!(kernel.parquet_data_file_cost_based_merged_range_plans, None);
528        assert_eq!(kernel.parquet_data_file_store_delegated_range_plans, None);
529        assert_eq!(kernel.parquet_data_file_range_get_operations, None);
530        assert_eq!(kernel.parquet_data_file_full_get_operations, None);
531        assert_eq!(kernel.parquet_data_file_bytes_received, None);
532        assert_eq!(kernel.estimated_parquet_task_bytes_admitted, None);
533    }
534
535    #[test]
536    fn debug_output_is_safe_and_redacted() {
537        assert_eq!(
538            format!("{:?}", metrics(ParquetReaderBackend::Direct)),
539            "DeltaScanMetrics { .. }"
540        );
541    }
542
543    #[test]
544    fn snapshot_maps_live_counters() {
545        let metrics = metrics(ParquetReaderBackend::Direct);
546        metrics.record_scan_partitions_planned(16);
547        let counters = [
548            &metrics.inner.scan_partitions_started,
549            &metrics.inner.scan_partitions_completed,
550            &metrics.inner.file_tasks_started,
551            &metrics.inner.file_tasks_completed,
552            &metrics.inner.scheduler_batches_emitted,
553            &metrics.inner.scheduler_rows_emitted,
554            &metrics.inner.deletion_vector_payloads_loaded,
555            &metrics.inner.deletion_vectors_applied,
556            &metrics.inner.deletion_vector_rows_deleted,
557            &metrics.inner.deletion_vector_failures,
558            &metrics.inner.deletion_vector_coordinate_rejections,
559            &metrics.inner.parquet_data_file_exact_ranges_requested,
560            &metrics.inner.parquet_data_file_exact_range_bytes_requested,
561            &metrics
562                .inner
563                .parquet_data_file_physical_range_requests_planned,
564            &metrics.inner.parquet_data_file_physical_range_bytes_planned,
565            &metrics.inner.parquet_data_file_cold_start_range_plans,
566            &metrics.inner.parquet_data_file_cost_based_exact_range_plans,
567            &metrics
568                .inner
569                .parquet_data_file_cost_based_merged_range_plans,
570            &metrics.inner.parquet_data_file_store_delegated_range_plans,
571            &metrics.inner.parquet_data_file_range_get_operations,
572            &metrics.inner.parquet_data_file_full_get_operations,
573            &metrics.inner.parquet_data_file_bytes_received,
574            &metrics.inner.estimated_parquet_task_bytes_admitted,
575        ];
576        for (index, counter) in counters.into_iter().enumerate() {
577            saturating_fetch_add(counter, u64::try_from(index + 1).expect("small test value"));
578        }
579
580        let snapshot = metrics.snapshot();
581        assert_eq!(snapshot.scan_partitions_planned, 16);
582        assert_eq!(snapshot.scan_partitions_started, 1);
583        assert_eq!(snapshot.scan_partitions_completed, 2);
584        assert_eq!(snapshot.file_tasks_started, 3);
585        assert_eq!(snapshot.file_tasks_completed, 4);
586        assert_eq!(snapshot.scheduler_batches_emitted, 5);
587        assert_eq!(snapshot.scheduler_rows_emitted, 6);
588        assert_eq!(snapshot.deletion_vector_payloads_loaded, 7);
589        assert_eq!(snapshot.deletion_vectors_applied, 8);
590        assert_eq!(snapshot.deletion_vector_rows_deleted, 9);
591        assert_eq!(snapshot.deletion_vector_failures, 10);
592        assert_eq!(snapshot.deletion_vector_coordinate_rejections, 11);
593        assert_eq!(snapshot.parquet_data_file_exact_ranges_requested, Some(12));
594        assert_eq!(
595            snapshot.parquet_data_file_exact_range_bytes_requested,
596            Some(13)
597        );
598        assert_eq!(
599            snapshot.parquet_data_file_physical_range_requests_planned,
600            Some(14)
601        );
602        assert_eq!(
603            snapshot.parquet_data_file_physical_range_bytes_planned,
604            Some(15)
605        );
606        assert_eq!(snapshot.parquet_data_file_cold_start_range_plans, Some(16));
607        assert_eq!(
608            snapshot.parquet_data_file_cost_based_exact_range_plans,
609            Some(17)
610        );
611        assert_eq!(
612            snapshot.parquet_data_file_cost_based_merged_range_plans,
613            Some(18)
614        );
615        assert_eq!(
616            snapshot.parquet_data_file_store_delegated_range_plans,
617            Some(19)
618        );
619        assert_eq!(snapshot.parquet_data_file_range_get_operations, Some(20));
620        assert_eq!(snapshot.parquet_data_file_full_get_operations, Some(21));
621        assert_eq!(snapshot.parquet_data_file_bytes_received, Some(22));
622        assert_eq!(snapshot.estimated_parquet_task_bytes_admitted, Some(23));
623    }
624
625    #[test]
626    fn cloned_handles_saturate_under_concurrent_updates() -> Result<(), &'static str> {
627        let metrics = metrics(ParquetReaderBackend::Direct);
628        metrics
629            .inner
630            .file_tasks_started
631            .store(u64::MAX - 1, Ordering::Relaxed);
632        let workers = (0..4)
633            .map(|_| {
634                let metrics = metrics.clone();
635                thread::spawn(move || {
636                    saturating_fetch_add(&metrics.inner.file_tasks_started, 1);
637                })
638            })
639            .collect::<Vec<_>>();
640
641        for worker in workers {
642            worker.join().map_err(|_| "metrics worker panicked")?;
643        }
644
645        assert_eq!(metrics.snapshot().file_tasks_started, u64::MAX);
646        Ok(())
647    }
648}