liquid-cache 0.1.12

10x lower latency for cloud-native DataFusion
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

mod internal_tracing;
mod stats;
mod tracer;

pub use internal_tracing::EventTrace;
pub use stats::{CacheStats, RuntimeStats, RuntimeStatsSnapshot};
pub use tracer::CacheTracer;

use std::path::Path;

use internal_tracing::EventTracer;
pub(crate) use internal_tracing::InternalEvent;
use stats::{RuntimeStats as RuntimeStatsInner, RuntimeStatsSnapshot as RuntimeStatsSnapshotInner};

#[derive(Debug)]
/// Cache-side observer for runtime stats, debug traces, and optional cache tracing.
pub struct Observer {
    runtime: RuntimeStatsInner,
    event_tracer: EventTracer,
    cache_tracer: CacheTracer,
}

impl Default for Observer {
    fn default() -> Self {
        Self::new()
    }
}

impl Observer {
    /// Create a new observer with all counters and traces reset.
    pub fn new() -> Self {
        Self {
            runtime: RuntimeStatsInner::default(),
            event_tracer: EventTracer::new(),
            cache_tracer: CacheTracer::new(),
        }
    }

    /// Snapshot runtime counters and reset them to zero.
    pub fn runtime_snapshot(&self) -> RuntimeStatsSnapshotInner {
        self.runtime.consume_snapshot()
    }

    /// Consume and clear the in-memory debug event trace.
    pub fn consume_event_trace(&self) -> EventTrace {
        self.event_tracer.drain()
    }

    /// Enable recording cache trace events (for offline analysis).
    pub fn enable_cache_trace(&self) {
        self.cache_tracer.enable();
    }

    /// Disable recording cache trace events.
    pub fn disable_cache_trace(&self) {
        self.cache_tracer.disable();
    }

    /// Flush recorded cache trace events to a Parquet file.
    pub fn flush_cache_trace(&self, to_file: impl AsRef<Path>) {
        self.cache_tracer.flush(to_file);
    }

    /// Access the underlying cache tracer.
    pub fn cache_tracer(&self) -> &CacheTracer {
        &self.cache_tracer
    }

    #[inline]
    pub(crate) fn on_get(&self, selection: bool) {
        self.runtime.incr_get();
        if selection {
            self.runtime.incr_get_with_selection();
        }
    }

    #[inline]
    pub(crate) fn on_try_read_liquid(&self) {
        self.runtime.incr_try_read_liquid();
    }

    #[inline]
    pub(crate) fn on_eval_predicate(&self) {
        self.runtime.incr_eval_predicate();
    }

    #[inline]
    pub(crate) fn on_get_squeezed_success(&self) {
        self.runtime.incr_get_squeezed_success();
    }

    #[inline]
    pub(crate) fn on_get_squeezed_needs_io(&self) {
        self.runtime.incr_get_squeezed_needs_io();
    }

    #[inline]
    pub(crate) fn on_hit_date32_expression(&self) {
        self.runtime.incr_hit_date32_expression();
    }

    pub(crate) fn record_internal(&self, event: InternalEvent) {
        match event {
            InternalEvent::IoWrite { .. } => self.runtime.incr_write_io_count(),
            InternalEvent::IoReadArrow { .. } | InternalEvent::IoReadLiquid { .. } => {
                self.runtime.incr_read_io_count()
            }
            InternalEvent::IoReadSqueezedBacking { .. } => {
                self.runtime.incr_read_io_count();
                self.runtime.incr_get_squeezed_needs_io();
            }
            InternalEvent::DecompressSqueezed {
                decompressed,
                total,
                ..
            } => {
                self.runtime
                    .track_decompress_squeezed_count(decompressed, total);
            }
            _ => {}
        }

        if cfg!(debug_assertions) {
            self.event_tracer.record(event);
        }
    }

    pub(crate) fn runtime_stats(&self) -> &RuntimeStats {
        &self.runtime
    }
}