llkv-table 0.8.5-alpha

Columnar table using the LLKV toolkit.
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

//! Table-level pager diagnostics built on top of pager statistics.
//!
//! - [`llkv_storage::pager::PagerDiagnostics`] captures raw [`IoStatsSnapshot`] values.
//! - This module wraps that handle so table loaders can bracket ingest work with
//!   [`TablePagerIngestionDiagnostics::begin_table`] / [`TablePagerIngestionDiagnostics::finish_table`] and receive
//!   a [`TablePagerIngestionSample`] (table-scoped snapshot delta).
//! - Consumers format `TablePagerIngestionSample` for reporting, while [`TablePagerIngestionDiagnostics`]
//!   keeps the mutex bookkeeping out of the application layer.

use llkv_storage::pager::{IoStatsSnapshot, PagerDiagnostics};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;

/// Captures pager I/O metrics for a single table ingest span.
#[derive(Clone, Debug)]
pub struct TablePagerIngestionSample {
    /// Logical table name for the recorded span.
    pub table: String,
    /// Rows written while the table ingest was active.
    pub rows: usize,
    /// Wall-clock duration of the ingest span.
    pub elapsed: Duration,
    /// Pager I/O deltas collected during the ingest span.
    pub delta: IoStatsSnapshot,
}

impl TablePagerIngestionSample {
    /// Overwrite percentage (0-100) derived from [`IoStatsSnapshot::overwritten_put_bytes`].
    pub fn overwrite_pct(&self) -> f64 {
        self.delta.overwrite_pct()
    }

    /// Average physical put operations per batch for this table ingest.
    pub fn puts_per_batch(&self) -> f64 {
        self.delta.puts_per_batch()
    }

    /// Average physical get operations per batch for this table ingest.
    pub fn gets_per_batch(&self) -> f64 {
        self.delta.gets_per_batch()
    }

    /// Fresh bytes written during this ingest converted to mebibytes.
    pub fn fresh_mib(&self) -> f64 {
        self.delta.fresh_mib()
    }

    /// Overwrite bytes written during this ingest converted to mebibytes.
    pub fn overwrite_mib(&self) -> f64 {
        self.delta.overwrite_mib()
    }
}

/// Tracks per-table pager spans over the course of a loader run.
#[derive(Debug)]
pub struct TablePagerIngestionDiagnostics {
    pager: Arc<PagerDiagnostics>,
    table_starts: Mutex<HashMap<String, IoStatsSnapshot>>,
    completed: Mutex<Vec<TablePagerIngestionSample>>,
}

impl TablePagerIngestionDiagnostics {
    /// Create diagnostics bound to the provided pager statistics helper.
    pub fn new(pager: Arc<PagerDiagnostics>) -> Self {
        Self {
            pager,
            table_starts: Mutex::new(HashMap::new()),
            completed: Mutex::new(Vec::new()),
        }
    }

    /// Record the start snapshot for a table ingest span.
    pub fn begin_table(&self, table: impl Into<String>) {
        let mut starts = self.table_starts.lock().unwrap();
        starts.insert(table.into(), self.pager.snapshot());
    }

    /// Complete a table ingest span and store the resulting diagnostics entry.
    pub fn finish_table(
        &self,
        table: &str,
        rows: usize,
        elapsed: Duration,
    ) -> TablePagerIngestionSample {
        let start_snapshot = {
            let mut starts = self.table_starts.lock().unwrap();
            starts
                .remove(table)
                .unwrap_or_else(|| self.pager.snapshot())
        };
        let delta = self.pager.delta_since(&start_snapshot);
        let entry = TablePagerIngestionSample {
            table: table.to_string(),
            rows,
            elapsed,
            delta,
        };
        self.completed.lock().unwrap().push(entry.clone());
        entry
    }

    /// Inspect the completed per-table diagnostics.
    pub fn completed_tables(&self) -> Vec<TablePagerIngestionSample> {
        self.completed.lock().unwrap().clone()
    }

    /// Return the cumulative pager totals captured during the run.
    pub fn totals(&self) -> IoStatsSnapshot {
        self.pager.totals()
    }
}