moraine 0.5.2

A SlateDB-backed catalog for DuckLake
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

//! Durable, bounded maintenance-pass status.

use slatedb::ErrorKind;

use crate::{
    Catalog, MaintenanceStatusPass, MaintenanceStatusStep, ReadOnlyCatalog, Timestamp,
    error::{Error, Result},
    store::{
        handle::ReadHandle,
        key::{Key, SysKey},
        proto::{MaintenanceStatusPassValue, MaintenanceStatusStepValue},
        read, value,
    },
    transaction::commit::{self, FORMAT_WITH_MAINTENANCE_STATUS, MAX_COMMIT_ATTEMPTS, StagedWrite},
};

const RETAINED_PASSES: usize = 16;

/// Reads the durable history, newest first.
pub(crate) async fn read(catalog: &ReadOnlyCatalog) -> Result<Vec<MaintenanceStatusPass>> {
    let session = catalog.begin_read().await?;
    let stored = read::read_maintenance_status(session.handle()).await;
    session.finish();

    Ok(stored?
        .unwrap_or_default()
        .passes
        .into_iter()
        .rev()
        .map(decode_pass)
        .collect())
}

/// Appends one pass and durably retains the newest bounded window.
pub(crate) async fn record(catalog: &Catalog, pass: MaintenanceStatusPass) -> Result<()> {
    let encoded = encode_pass(pass);

    for attempt in 0..MAX_COMMIT_ATTEMPTS {
        if attempt > 0 {
            tokio::time::sleep(commit::retry_backoff(attempt)).await;
        }

        let tx = catalog.begin_write_tx().await?;
        let mut status = read::read_maintenance_status(ReadHandle::Tx(&tx))
            .await?
            .unwrap_or_default();
        status.passes.push(encoded.clone());
        if status.passes.len() > RETAINED_PASSES {
            let remove = status.passes.len() - RETAINED_PASSES;
            status.passes.drain(..remove);
        }

        let mut writes: Vec<StagedWrite> = vec![(
            Key::Sys(SysKey::MaintenanceStatus).encode(),
            Some(value::encode_value(&status)),
        )];
        if let Some(stamp) = commit::format_stamp_to(&tx, FORMAT_WITH_MAINTENANCE_STATUS).await? {
            writes.push(stamp);
        }
        let staged = commit::stage_writes(&tx, &writes)?;

        match commit::commit_durable(tx, "maintenance status", staged).await {
            Ok(_) => return Ok(()),
            Err(error) if error.kind() == ErrorKind::Transaction => {}
            Err(error) => return Err(Error::from(error)),
        }
    }

    Err(Error::RetryBudgetExhausted(format!(
        "maintenance status did not settle in {MAX_COMMIT_ATTEMPTS} attempts"
    )))
}

fn encode_pass(pass: MaintenanceStatusPass) -> MaintenanceStatusPassValue {
    MaintenanceStatusPassValue {
        started_at_micros: pass.started_at.as_micros(),
        trigger: pass.trigger,
        steps: pass
            .steps
            .into_iter()
            .map(|step| MaintenanceStatusStepValue {
                step: step.step,
                status: step.status,
                detail: step.detail,
            })
            .collect(),
    }
}

fn decode_pass(pass: MaintenanceStatusPassValue) -> MaintenanceStatusPass {
    MaintenanceStatusPass::new(
        Timestamp::from_micros(pass.started_at_micros),
        pass.trigger,
        pass.steps
            .into_iter()
            .map(|step| MaintenanceStatusStep::new(step.step, step.status, step.detail))
            .collect(),
    )
}