buoyant_kernel 0.21.101

Buoyant Data distribution of delta-kernel
Documentation 
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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151

//! Metrics for scan log replay operations.

use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::time::Duration;

use tracing::info;

use crate::metrics::{MetricEvent, MetricId, ScanType};

/// Metrics collected during scan log replay. Metrics are updated and read using relaxed ordering
/// to keep updates fast across parallel executing threads.
pub(crate) struct ScanMetrics {
    /// Add files seen during add remove deduplication. This does not include data skipped add
    /// files.
    /// Java equivalent: `addFilesCounter`
    num_add_files_seen: AtomicU64,
    /// Add files that survived log replay (files to read). includes files that survived
    /// dataskipping, partition pruning, and add/remove deduplication.
    /// Java equivalent: `activeAddFilesCounter`
    num_active_add_files: AtomicU64,
    /// Remove files seen (from delta/commit files only).
    /// Java equivalent: `removeFilesFromDeltaFilesCounter`
    num_remove_files_seen: AtomicU64,
    /// Non-file actions seen (protocol, metadata, etc.).
    num_non_file_actions: AtomicU64,
    /// Files filtered by predicates (data skipping + partition pruning).
    num_predicate_filtered: AtomicU64,
    /// Peak size of the deduplication hash set.
    peak_hash_set_size: AtomicUsize,
    /// Time spent in the deduplication visitor (ns).
    dedup_visitor_time_ns: AtomicU64,
    /// Time spent evaluating predicates (ns). This includes data skipping and partition pruning.
    predicate_eval_time_ns: AtomicU64,
}

impl Default for ScanMetrics {
    fn default() -> Self {
        Self {
            num_add_files_seen: AtomicU64::new(0),
            num_active_add_files: AtomicU64::new(0),
            num_remove_files_seen: AtomicU64::new(0),
            num_non_file_actions: AtomicU64::new(0),
            num_predicate_filtered: AtomicU64::new(0),
            peak_hash_set_size: AtomicUsize::new(0),
            dedup_visitor_time_ns: AtomicU64::new(0),
            predicate_eval_time_ns: AtomicU64::new(0),
        }
    }
}

impl ScanMetrics {
    pub(crate) fn incr_add_files_seen(&self) {
        self.num_add_files_seen.fetch_add(1, Ordering::Relaxed);
    }

    pub(crate) fn incr_active_add_files(&self) {
        self.num_active_add_files.fetch_add(1, Ordering::Relaxed);
    }

    pub(crate) fn incr_remove_files_seen(&self) {
        self.num_remove_files_seen.fetch_add(1, Ordering::Relaxed);
    }

    pub(crate) fn incr_non_file_actions(&self) {
        self.num_non_file_actions.fetch_add(1, Ordering::Relaxed);
    }

    pub(crate) fn add_predicate_filtered(&self, value: u64) {
        self.num_predicate_filtered
            .fetch_add(value, Ordering::Relaxed);
    }

    pub(crate) fn update_peak_hash_set_size(&self, value: usize) {
        self.peak_hash_set_size.fetch_max(value, Ordering::Relaxed);
    }

    pub(crate) fn add_dedup_visitor_time_ns(&self, duration_ns: u64) {
        self.dedup_visitor_time_ns
            .fetch_add(duration_ns, Ordering::Relaxed);
    }

    pub(crate) fn add_predicate_eval_time_ns(&self, duration_ns: u64) {
        self.predicate_eval_time_ns
            .fetch_add(duration_ns, Ordering::Relaxed);
    }

    /// Reset counters to zero for a new phase.
    ///
    /// This is used between sequential and parallel phases to get fresh metrics
    /// without reconstructing the entire processor. The peak hash set size is
    /// preserved since it represents a high-water mark across all phases.
    pub(crate) fn reset_counters(&self) {
        self.num_add_files_seen.store(0, Ordering::Relaxed);
        self.num_active_add_files.store(0, Ordering::Relaxed);
        self.num_remove_files_seen.store(0, Ordering::Relaxed);
        self.num_non_file_actions.store(0, Ordering::Relaxed);
        self.num_predicate_filtered.store(0, Ordering::Relaxed);
        self.dedup_visitor_time_ns.store(0, Ordering::Relaxed);
        self.predicate_eval_time_ns.store(0, Ordering::Relaxed);
    }

    /// Snapshot all counters into a `MetricEvent::ScanMetadataCompleted`.
    ///
    /// `scan_type` identifies whether this event was emitted by full scan metadata replay or
    /// by a phase of parallel scan metadata replay.
    pub(crate) fn to_event(
        &self,
        operation_id: MetricId,
        scan_type: ScanType,
        total_duration: Duration,
    ) -> MetricEvent {
        MetricEvent::ScanMetadataCompleted {
            operation_id,
            scan_type,
            total_duration,
            num_add_files_seen: self.num_add_files_seen.load(Ordering::Relaxed),
            num_active_add_files: self.num_active_add_files.load(Ordering::Relaxed),
            num_remove_files_seen: self.num_remove_files_seen.load(Ordering::Relaxed),
            num_non_file_actions: self.num_non_file_actions.load(Ordering::Relaxed),
            num_predicate_filtered: self.num_predicate_filtered.load(Ordering::Relaxed),
            peak_hash_set_size: self.peak_hash_set_size.load(Ordering::Relaxed),
            dedup_visitor_time_ms: self.dedup_visitor_time_ns.load(Ordering::Relaxed) / 1_000_000,
            predicate_eval_time_ms: self.predicate_eval_time_ns.load(Ordering::Relaxed) / 1_000_000,
        }
    }

    /// Log all metrics with a message in the current tracing span context.
    pub(crate) fn log(&self, message: impl AsRef<str>) {
        let add_files_seen = self.num_add_files_seen.load(Ordering::Relaxed);
        let active_add_files = self.num_active_add_files.load(Ordering::Relaxed);
        let remove_files_seen = self.num_remove_files_seen.load(Ordering::Relaxed);
        let non_file_actions = self.num_non_file_actions.load(Ordering::Relaxed);
        let predicate_filtered = self.num_predicate_filtered.load(Ordering::Relaxed);
        let peak_hash_set_size = self.peak_hash_set_size.load(Ordering::Relaxed);
        let dedup_visitor_time_ms = self.dedup_visitor_time_ns.load(Ordering::Relaxed) / 1_000_000;
        let predicate_eval_time_ms =
            self.predicate_eval_time_ns.load(Ordering::Relaxed) / 1_000_000;
        info!(
            add_files_seen,
            active_add_files,
            remove_files_seen,
            non_file_actions,
            predicate_filtered,
            peak_hash_set_size,
            dedup_visitor_time_ms,
            predicate_eval_time_ms,
            "{}",
            message.as_ref()
        );
    }
}