delta_funnel/query_engine/datafusion/execution/
read_stats.rs

1//! Provider read progress counters for Delta scan execution.
2//!
3//! These counters are intentionally independent from DataFusion's metrics set.
4//! They are the provider-owned handoff for later orchestration code that needs
5//! partial progress after success, failure, or cancellation.
6
7use std::sync::atomic::{AtomicU64, Ordering};
8
9use super::scheduling::DeltaProviderReaderBackend;
10
11/// Immutable view of provider read progress for one physical scan.
12#[allow(dead_code)]
13#[derive(Debug, Clone, PartialEq, Eq)]
14pub struct DeltaProviderReadStatsSnapshot {
15    /// DataFusion table name for this source.
16    pub source_name: String,
17    /// Delta snapshot version selected for this scan.
18    pub snapshot_version: u64,
19    /// Provider file-reader backend selected for this scan.
20    pub reader_backend: DeltaProviderReaderBackend,
21    /// Whether metadata expansion exhausted the upstream kernel scan iterator.
22    pub scan_metadata_exhausted: Option<bool>,
23    /// Planned DataFusion execution partitions for this scan.
24    pub scan_partitions_planned: u64,
25    /// Selected provider file tasks planned for this scan.
26    pub files_planned: u64,
27    /// Estimated output rows from planning when every selected file had stats.
28    pub estimated_rows: Option<u64>,
29    /// Estimated bytes from planning when every selected file had a byte size.
30    pub estimated_bytes: Option<u64>,
31    /// Effective DataFusion task batch size observed when execution starts.
32    ///
33    /// This records the upstream query-engine setting. It does not claim every
34    /// produced or final query-output batch has exactly this many rows.
35    pub datafusion_output_batch_size: Option<u64>,
36    /// Execution partitions whose stream was started by DataFusion.
37    pub scan_partitions_started: u64,
38    /// Execution partitions whose stream reached normal completion.
39    pub scan_partitions_completed: u64,
40    /// File-read handoffs that were started.
41    pub files_started: u64,
42    /// File-read handoffs that finished successfully.
43    pub files_completed: u64,
44    /// File tasks skipped before read scheduling by dynamic partition pruning.
45    pub dynamic_partition_files_pruned: u64,
46    /// File tasks kept after dynamic partition pruning evaluation.
47    pub dynamic_partition_files_kept: u64,
48    /// Post-phase physical filters offered to the Delta dynamic filter hook.
49    pub dynamic_filters_received: u64,
50    /// Offered dynamic filters retained for partition pruning.
51    pub dynamic_filters_accepted: u64,
52    /// Offered filters rejected by the dynamic filter hook policy.
53    pub dynamic_filters_unsupported: u64,
54    /// Attempts to snapshot a retained dynamic filter during file admission.
55    pub dynamic_filter_snapshots: u64,
56    /// Kept file tasks with missing, invalid, or unparsable partition metadata.
57    pub dynamic_partition_files_not_pruned_missing_metadata: u64,
58    /// Kept file tasks with unsupported or failed dynamic partition evaluation.
59    pub dynamic_partition_files_not_pruned_unsupported_expression: u64,
60    /// Record batches sent toward DataFusion.
61    pub batches_produced: u64,
62    /// Rows sent toward DataFusion after transform and DV filtering.
63    pub rows_produced: u64,
64    /// Deletion-vector payloads loaded for selected files.
65    pub deletion_vector_payloads_loaded: u64,
66    /// Deletion-vector masks applied to selected files.
67    pub deletion_vectors_applied: u64,
68    /// Rows removed by deletion-vector masks when known.
69    pub deletion_vector_rows_deleted: u64,
70    /// Deletion-vector read or masking failures.
71    pub deletion_vector_failures: u64,
72    /// Deletion-vector reads rejected by safety gates.
73    pub deletion_vector_rejections: u64,
74}
75
76/// Static context and planning estimates for one provider read stats instance.
77#[allow(dead_code)]
78pub(crate) struct DeltaProviderReadStatsConfig {
79    /// DataFusion table name for this source.
80    pub(crate) source_name: String,
81    /// Delta snapshot version selected for this scan.
82    pub(crate) snapshot_version: u64,
83    /// Provider file-reader backend selected for this scan.
84    pub(crate) reader_backend: DeltaProviderReaderBackend,
85    /// Whether metadata expansion exhausted the upstream kernel scan iterator.
86    pub(crate) scan_metadata_exhausted: Option<bool>,
87    /// Planned DataFusion execution partitions for this scan.
88    pub(crate) scan_partitions_planned: usize,
89    /// Selected provider file tasks planned for this scan.
90    pub(crate) files_planned: usize,
91    /// Estimated output rows from planning when every selected file had stats.
92    pub(crate) estimated_rows: Option<u64>,
93    /// Estimated bytes from planning when every selected file had a byte size.
94    pub(crate) estimated_bytes: Option<u64>,
95}
96
97/// Thread-safe provider read progress for one physical scan.
98#[allow(dead_code)]
99#[derive(Debug)]
100pub(crate) struct DeltaProviderReadStats {
101    source_name: String,
102    snapshot_version: u64,
103    reader_backend: DeltaProviderReaderBackend,
104    scan_metadata_exhausted: Option<bool>,
105    scan_partitions_planned: u64,
106    files_planned: u64,
107    estimated_rows: Option<u64>,
108    estimated_bytes: Option<u64>,
109    datafusion_output_batch_size: AtomicU64,
110    scan_partitions_started: AtomicU64,
111    scan_partitions_completed: AtomicU64,
112    files_started: AtomicU64,
113    files_completed: AtomicU64,
114    dynamic_partition_files_pruned: AtomicU64,
115    dynamic_partition_files_kept: AtomicU64,
116    dynamic_filters_received: AtomicU64,
117    dynamic_filters_accepted: AtomicU64,
118    dynamic_filters_unsupported: AtomicU64,
119    dynamic_filter_snapshots: AtomicU64,
120    dynamic_partition_files_not_pruned_missing_metadata: AtomicU64,
121    dynamic_partition_files_not_pruned_unsupported_expression: AtomicU64,
122    batches_produced: AtomicU64,
123    rows_produced: AtomicU64,
124    deletion_vector_payloads_loaded: AtomicU64,
125    deletion_vectors_applied: AtomicU64,
126    deletion_vector_rows_deleted: AtomicU64,
127    deletion_vector_failures: AtomicU64,
128    deletion_vector_rejections: AtomicU64,
129}
130
131#[allow(dead_code)]
132impl DeltaProviderReadStats {
133    /// Creates zeroed read progress for one physical scan.
134    #[allow(dead_code)]
135    #[must_use]
136    pub(crate) fn new(config: DeltaProviderReadStatsConfig) -> Self {
137        Self {
138            source_name: config.source_name,
139            snapshot_version: config.snapshot_version,
140            reader_backend: config.reader_backend,
141            scan_metadata_exhausted: config.scan_metadata_exhausted,
142            scan_partitions_planned: usize_to_u64_saturating(config.scan_partitions_planned),
143            files_planned: usize_to_u64_saturating(config.files_planned),
144            estimated_rows: config.estimated_rows,
145            estimated_bytes: config.estimated_bytes,
146            datafusion_output_batch_size: AtomicU64::new(0),
147            scan_partitions_started: AtomicU64::new(0),
148            scan_partitions_completed: AtomicU64::new(0),
149            files_started: AtomicU64::new(0),
150            files_completed: AtomicU64::new(0),
151            dynamic_partition_files_pruned: AtomicU64::new(0),
152            dynamic_partition_files_kept: AtomicU64::new(0),
153            dynamic_filters_received: AtomicU64::new(0),
154            dynamic_filters_accepted: AtomicU64::new(0),
155            dynamic_filters_unsupported: AtomicU64::new(0),
156            dynamic_filter_snapshots: AtomicU64::new(0),
157            dynamic_partition_files_not_pruned_missing_metadata: AtomicU64::new(0),
158            dynamic_partition_files_not_pruned_unsupported_expression: AtomicU64::new(0),
159            batches_produced: AtomicU64::new(0),
160            rows_produced: AtomicU64::new(0),
161            deletion_vector_payloads_loaded: AtomicU64::new(0),
162            deletion_vectors_applied: AtomicU64::new(0),
163            deletion_vector_rows_deleted: AtomicU64::new(0),
164            deletion_vector_failures: AtomicU64::new(0),
165            deletion_vector_rejections: AtomicU64::new(0),
166        }
167    }
168
169    /// Returns a point-in-time copy of all counters.
170    #[allow(dead_code)]
171    #[must_use]
172    pub(crate) fn snapshot(&self) -> DeltaProviderReadStatsSnapshot {
173        DeltaProviderReadStatsSnapshot {
174            source_name: self.source_name.clone(),
175            snapshot_version: self.snapshot_version,
176            reader_backend: self.reader_backend,
177            scan_metadata_exhausted: self.scan_metadata_exhausted,
178            scan_partitions_planned: self.scan_partitions_planned,
179            files_planned: self.files_planned,
180            estimated_rows: self.estimated_rows,
181            estimated_bytes: self.estimated_bytes,
182            datafusion_output_batch_size: nonzero_atomic_snapshot(
183                &self.datafusion_output_batch_size,
184            ),
185            scan_partitions_started: self.scan_partitions_started.load(Ordering::Relaxed),
186            scan_partitions_completed: self.scan_partitions_completed.load(Ordering::Relaxed),
187            files_started: self.files_started.load(Ordering::Relaxed),
188            files_completed: self.files_completed.load(Ordering::Relaxed),
189            dynamic_partition_files_pruned: self
190                .dynamic_partition_files_pruned
191                .load(Ordering::Relaxed),
192            dynamic_partition_files_kept: self.dynamic_partition_files_kept.load(Ordering::Relaxed),
193            dynamic_filters_received: self.dynamic_filters_received.load(Ordering::Relaxed),
194            dynamic_filters_accepted: self.dynamic_filters_accepted.load(Ordering::Relaxed),
195            dynamic_filters_unsupported: self.dynamic_filters_unsupported.load(Ordering::Relaxed),
196            dynamic_filter_snapshots: self.dynamic_filter_snapshots.load(Ordering::Relaxed),
197            dynamic_partition_files_not_pruned_missing_metadata: self
198                .dynamic_partition_files_not_pruned_missing_metadata
199                .load(Ordering::Relaxed),
200            dynamic_partition_files_not_pruned_unsupported_expression: self
201                .dynamic_partition_files_not_pruned_unsupported_expression
202                .load(Ordering::Relaxed),
203            batches_produced: self.batches_produced.load(Ordering::Relaxed),
204            rows_produced: self.rows_produced.load(Ordering::Relaxed),
205            deletion_vector_payloads_loaded: self
206                .deletion_vector_payloads_loaded
207                .load(Ordering::Relaxed),
208            deletion_vectors_applied: self.deletion_vectors_applied.load(Ordering::Relaxed),
209            deletion_vector_rows_deleted: self.deletion_vector_rows_deleted.load(Ordering::Relaxed),
210            deletion_vector_failures: self.deletion_vector_failures.load(Ordering::Relaxed),
211            deletion_vector_rejections: self.deletion_vector_rejections.load(Ordering::Relaxed),
212        }
213    }
214
215    pub(crate) fn record_scan_partition_started(&self) {
216        saturating_fetch_add(&self.scan_partitions_started, 1);
217    }
218
219    pub(crate) fn record_datafusion_output_batch_size(&self, batch_size: usize) {
220        self.datafusion_output_batch_size
221            .store(usize_to_u64_saturating(batch_size), Ordering::Relaxed);
222    }
223
224    pub(crate) fn record_scan_partition_completed(&self) {
225        saturating_fetch_add(&self.scan_partitions_completed, 1);
226    }
227
228    pub(crate) fn record_file_started(&self) {
229        saturating_fetch_add(&self.files_started, 1);
230    }
231
232    pub(crate) fn record_file_completed(&self) {
233        saturating_fetch_add(&self.files_completed, 1);
234    }
235
236    pub(crate) fn record_dynamic_partition_file_pruned(&self) {
237        saturating_fetch_add(&self.dynamic_partition_files_pruned, 1);
238    }
239
240    pub(crate) fn record_dynamic_partition_file_kept(&self) {
241        saturating_fetch_add(&self.dynamic_partition_files_kept, 1);
242    }
243
244    pub(crate) fn record_dynamic_filters_received(&self, count: usize) {
245        saturating_fetch_add(
246            &self.dynamic_filters_received,
247            usize_to_u64_saturating(count),
248        );
249    }
250
251    pub(crate) fn record_dynamic_filters_accepted(&self, count: usize) {
252        saturating_fetch_add(
253            &self.dynamic_filters_accepted,
254            usize_to_u64_saturating(count),
255        );
256    }
257
258    pub(crate) fn record_dynamic_filters_unsupported(&self, count: usize) {
259        saturating_fetch_add(
260            &self.dynamic_filters_unsupported,
261            usize_to_u64_saturating(count),
262        );
263    }
264
265    pub(crate) fn record_dynamic_filter_snapshot(&self) {
266        saturating_fetch_add(&self.dynamic_filter_snapshots, 1);
267    }
268
269    pub(crate) fn record_dynamic_partition_file_not_pruned_missing_metadata(&self) {
270        saturating_fetch_add(&self.dynamic_partition_files_not_pruned_missing_metadata, 1);
271    }
272
273    pub(crate) fn record_dynamic_partition_file_not_pruned_unsupported_expression(&self) {
274        saturating_fetch_add(
275            &self.dynamic_partition_files_not_pruned_unsupported_expression,
276            1,
277        );
278    }
279
280    pub(crate) fn record_batch_produced(&self, rows: usize) {
281        saturating_fetch_add(&self.batches_produced, 1);
282        saturating_fetch_add(&self.rows_produced, usize_to_u64_saturating(rows));
283    }
284
285    pub(crate) fn record_deletion_vector_payload_loaded(&self) {
286        saturating_fetch_add(&self.deletion_vector_payloads_loaded, 1);
287    }
288
289    pub(crate) fn record_deletion_vector_applied(&self, deleted_rows: usize) {
290        saturating_fetch_add(&self.deletion_vectors_applied, 1);
291        self.record_deletion_vector_rows_deleted(deleted_rows);
292    }
293
294    pub(crate) fn record_deletion_vector_rows_deleted(&self, deleted_rows: usize) {
295        saturating_fetch_add(
296            &self.deletion_vector_rows_deleted,
297            usize_to_u64_saturating(deleted_rows),
298        );
299    }
300
301    pub(crate) fn record_deletion_vector_failure(&self) {
302        saturating_fetch_add(&self.deletion_vector_failures, 1);
303    }
304
305    pub(crate) fn record_deletion_vector_rejection(&self) {
306        saturating_fetch_add(&self.deletion_vector_rejections, 1);
307    }
308}
309
310#[allow(dead_code)]
311fn saturating_fetch_add(counter: &AtomicU64, value: u64) {
312    let mut current = counter.load(Ordering::Relaxed);
313    loop {
314        let next = current.saturating_add(value);
315        match counter.compare_exchange_weak(current, next, Ordering::Relaxed, Ordering::Relaxed) {
316            Ok(_) => return,
317            Err(observed) => current = observed,
318        }
319    }
320}
321
322#[allow(dead_code)]
323fn usize_to_u64_saturating(value: usize) -> u64 {
324    u64::try_from(value).unwrap_or(u64::MAX)
325}
326
327#[allow(dead_code)]
328fn nonzero_atomic_snapshot(counter: &AtomicU64) -> Option<u64> {
329    match counter.load(Ordering::Relaxed) {
330        0 => None,
331        value => Some(value),
332    }
333}
334
335#[cfg(test)]
336mod tests {
337    use std::sync::Arc;
338    use std::thread;
339
340    use super::{DeltaProviderReadStats, DeltaProviderReadStatsConfig};
341    use crate::query_engine::datafusion::execution::DeltaProviderReaderBackend;
342
343    #[test]
344    fn read_stats_snapshot_starts_with_context_and_zero_counters() {
345        let stats = DeltaProviderReadStats::new(DeltaProviderReadStatsConfig {
346            source_name: "orders".to_owned(),
347            snapshot_version: 7,
348            reader_backend: DeltaProviderReaderBackend::OfficialKernel,
349            scan_metadata_exhausted: Some(true),
350            scan_partitions_planned: 3,
351            files_planned: 5,
352            estimated_rows: Some(99),
353            estimated_bytes: Some(42),
354        });
355        let snapshot = stats.snapshot();
356
357        assert_eq!(snapshot.source_name, "orders");
358        assert_eq!(snapshot.snapshot_version, 7);
359        assert_eq!(
360            snapshot.reader_backend,
361            DeltaProviderReaderBackend::OfficialKernel
362        );
363        assert_eq!(snapshot.scan_metadata_exhausted, Some(true));
364        assert_eq!(snapshot.scan_partitions_planned, 3);
365        assert_eq!(snapshot.files_planned, 5);
366        assert_eq!(snapshot.estimated_rows, Some(99));
367        assert_eq!(snapshot.estimated_bytes, Some(42));
368        assert_eq!(snapshot.datafusion_output_batch_size, None);
369        assert_eq!(snapshot.scan_partitions_started, 0);
370        assert_eq!(snapshot.scan_partitions_completed, 0);
371        assert_eq!(snapshot.files_started, 0);
372        assert_eq!(snapshot.files_completed, 0);
373        assert_eq!(snapshot.dynamic_partition_files_pruned, 0);
374        assert_eq!(snapshot.dynamic_partition_files_kept, 0);
375        assert_eq!(snapshot.dynamic_filters_received, 0);
376        assert_eq!(snapshot.dynamic_filters_accepted, 0);
377        assert_eq!(snapshot.dynamic_filters_unsupported, 0);
378        assert_eq!(snapshot.dynamic_filter_snapshots, 0);
379        assert_eq!(
380            snapshot.dynamic_partition_files_not_pruned_missing_metadata,
381            0
382        );
383        assert_eq!(
384            snapshot.dynamic_partition_files_not_pruned_unsupported_expression,
385            0
386        );
387        assert_eq!(snapshot.batches_produced, 0);
388        assert_eq!(snapshot.rows_produced, 0);
389        assert_eq!(snapshot.deletion_vector_payloads_loaded, 0);
390        assert_eq!(snapshot.deletion_vectors_applied, 0);
391        assert_eq!(snapshot.deletion_vector_rows_deleted, 0);
392        assert_eq!(snapshot.deletion_vector_failures, 0);
393        assert_eq!(snapshot.deletion_vector_rejections, 0);
394    }
395
396    #[test]
397    fn read_stats_records_partial_progress_without_completing_failed_work() {
398        let stats = DeltaProviderReadStats::new(DeltaProviderReadStatsConfig {
399            source_name: "orders".to_owned(),
400            snapshot_version: 7,
401            reader_backend: DeltaProviderReaderBackend::OfficialKernel,
402            scan_metadata_exhausted: Some(false),
403            scan_partitions_planned: 1,
404            files_planned: 1,
405            estimated_rows: None,
406            estimated_bytes: None,
407        });
408
409        stats.record_scan_partition_started();
410        stats.record_datafusion_output_batch_size(8192);
411        stats.record_file_started();
412        stats.record_dynamic_partition_file_pruned();
413        stats.record_dynamic_partition_file_kept();
414        stats.record_dynamic_filters_received(3);
415        stats.record_dynamic_filters_accepted(1);
416        stats.record_dynamic_filters_unsupported(2);
417        stats.record_dynamic_filter_snapshot();
418        stats.record_dynamic_partition_file_not_pruned_missing_metadata();
419        stats.record_dynamic_partition_file_not_pruned_unsupported_expression();
420        stats.record_batch_produced(3);
421        stats.record_deletion_vector_payload_loaded();
422        stats.record_deletion_vector_applied(1);
423        stats.record_deletion_vector_failure();
424
425        let snapshot = stats.snapshot();
426
427        assert_eq!(snapshot.datafusion_output_batch_size, Some(8192));
428        assert_eq!(snapshot.scan_partitions_started, 1);
429        assert_eq!(snapshot.scan_partitions_completed, 0);
430        assert_eq!(snapshot.files_started, 1);
431        assert_eq!(snapshot.files_completed, 0);
432        assert_eq!(snapshot.dynamic_partition_files_pruned, 1);
433        assert_eq!(snapshot.dynamic_partition_files_kept, 1);
434        assert_eq!(snapshot.dynamic_filters_received, 3);
435        assert_eq!(snapshot.dynamic_filters_accepted, 1);
436        assert_eq!(snapshot.dynamic_filters_unsupported, 2);
437        assert_eq!(snapshot.dynamic_filter_snapshots, 1);
438        assert_eq!(
439            snapshot.dynamic_partition_files_not_pruned_missing_metadata,
440            1
441        );
442        assert_eq!(
443            snapshot.dynamic_partition_files_not_pruned_unsupported_expression,
444            1
445        );
446        assert_eq!(snapshot.batches_produced, 1);
447        assert_eq!(snapshot.rows_produced, 3);
448        assert_eq!(snapshot.deletion_vector_payloads_loaded, 1);
449        assert_eq!(snapshot.deletion_vectors_applied, 1);
450        assert_eq!(snapshot.deletion_vector_rows_deleted, 1);
451        assert_eq!(snapshot.deletion_vector_failures, 1);
452        assert_eq!(snapshot.deletion_vector_rejections, 0);
453    }
454
455    #[test]
456    fn read_stats_updates_are_thread_safe() -> Result<(), Box<dyn std::error::Error>> {
457        const THREADS: usize = 4;
458        const ITERATIONS: usize = 100;
459
460        let stats = Arc::new(DeltaProviderReadStats::new(DeltaProviderReadStatsConfig {
461            source_name: "orders".to_owned(),
462            snapshot_version: 7,
463            reader_backend: DeltaProviderReaderBackend::OfficialKernel,
464            scan_metadata_exhausted: None,
465            scan_partitions_planned: THREADS,
466            files_planned: THREADS,
467            estimated_rows: None,
468            estimated_bytes: None,
469        }));
470        let mut handles = Vec::new();
471
472        for _ in 0..THREADS {
473            let stats = Arc::clone(&stats);
474            handles.push(thread::spawn(move || {
475                for _ in 0..ITERATIONS {
476                    stats.record_scan_partition_started();
477                    stats.record_scan_partition_completed();
478                    stats.record_file_started();
479                    stats.record_file_completed();
480                    stats.record_dynamic_partition_file_pruned();
481                    stats.record_dynamic_partition_file_kept();
482                    stats.record_dynamic_filters_received(3);
483                    stats.record_dynamic_filters_accepted(1);
484                    stats.record_dynamic_filters_unsupported(2);
485                    stats.record_dynamic_filter_snapshot();
486                    stats.record_dynamic_partition_file_not_pruned_missing_metadata();
487                    stats.record_dynamic_partition_file_not_pruned_unsupported_expression();
488                    stats.record_batch_produced(2);
489                    stats.record_deletion_vector_payload_loaded();
490                    stats.record_deletion_vector_applied(1);
491                    stats.record_deletion_vector_rejection();
492                }
493            }));
494        }
495
496        for handle in handles {
497            handle.join().map_err(|_| "stats worker panicked")?;
498        }
499
500        let snapshot = stats.snapshot();
501        let expected = u64::try_from(THREADS * ITERATIONS)?;
502
503        assert_eq!(snapshot.scan_metadata_exhausted, None);
504        assert_eq!(snapshot.scan_partitions_started, expected);
505        assert_eq!(snapshot.scan_partitions_completed, expected);
506        assert_eq!(snapshot.files_started, expected);
507        assert_eq!(snapshot.files_completed, expected);
508        assert_eq!(snapshot.dynamic_partition_files_pruned, expected);
509        assert_eq!(snapshot.dynamic_partition_files_kept, expected);
510        assert_eq!(snapshot.dynamic_filters_received, expected * 3);
511        assert_eq!(snapshot.dynamic_filters_accepted, expected);
512        assert_eq!(snapshot.dynamic_filters_unsupported, expected * 2);
513        assert_eq!(snapshot.dynamic_filter_snapshots, expected);
514        assert_eq!(
515            snapshot.dynamic_partition_files_not_pruned_missing_metadata,
516            expected
517        );
518        assert_eq!(
519            snapshot.dynamic_partition_files_not_pruned_unsupported_expression,
520            expected
521        );
522        assert_eq!(snapshot.batches_produced, expected);
523        assert_eq!(snapshot.rows_produced, expected * 2);
524        assert_eq!(snapshot.deletion_vector_payloads_loaded, expected);
525        assert_eq!(snapshot.deletion_vectors_applied, expected);
526        assert_eq!(snapshot.deletion_vector_rows_deleted, expected);
527        assert_eq!(snapshot.deletion_vector_rejections, expected);
528
529        Ok(())
530    }
531}
delta_funnel/query_engine/datafusion/execution/read_stats.rs

delta_funnel/query_engine/datafusion/execution/
read_stats.rs
