moraine 0.4.3

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
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

//! Folding a committed batch forward into a materialized view.
//!
//! The store after a commit is the store before it plus the batch's writes,
//! so applying those same writes to the pre-commit [`CatalogSnapshot`] — put
//! the decoded value at each `current` key, remove it on a delete — yields
//! exactly what a fresh scan of the new head would build. That lets a writer
//! reuse the view it already holds instead of rescanning on every commit.
//!
//! Deletes mirror the store key by key with no cascade: a fresh scan simply
//! omits a deleted key, so dropping a table here must not also drop its
//! columns — the batch carries their own deletes when they end.

use super::StagedWrite;
use crate::{
    catalog::CatalogSnapshot,
    error::{Error, Result},
    store::{
        key::{CurrentKey, EntityKey, Key, SysKey},
        proto::{HeadValue, SnapshotValue},
        read::decode_entity,
        value,
    },
};

/// Folds one committed batch into `view`, advancing it to the batch's head.
/// A write whose key cannot be decoded, or a deletion of an immutable kind,
/// is refused so the caller drops the cached view rather than serve a view
/// that has silently diverged from the store.
pub(crate) fn fold_batch(view: &mut CatalogSnapshot, writes: &[StagedWrite]) -> Result<()> {
    let mut new_head: Option<u64> = None;
    let mut minted: Option<SnapshotValue> = None;

    for (encoded_key, write) in writes {
        let bytes = write.as_deref();
        match Key::decode(encoded_key)? {
            Key::Current(CurrentKey::Entity(entity)) => match bytes {
                Some(bytes) => view.put_record(decode_entity(entity, bytes)?),
                None => remove_entity(view, entity)?,
            },
            Key::Current(CurrentKey::GcFile { data_file_id }) => match bytes {
                Some(bytes) => view.put_gc_file(value::decode_value(bytes)?),
                None => view.remove_gc_file(data_file_id),
            },
            // The minted snapshot record; an expiry deletes older ones,
            // which leave the head view's entities untouched.
            Key::Snapshot { .. } => {
                if let Some(bytes) = bytes {
                    minted = Some(value::decode_value(bytes)?);
                }
            }
            Key::Sys(SysKey::Head) => {
                if let Some(bytes) = bytes {
                    let head: HeadValue = value::decode_value(bytes)?;
                    new_head = Some(head.snapshot_id);
                    view.batch_seq = head.batch_seq;
                }
            }
            // History mirrors, index entries, inline data, and the format
            // stamp do not appear in the current head view.
            _ => {}
        }
    }

    // Adopt the minted snapshot's metadata only when it is the new head;
    // a head-preserving (maintenance) batch keeps the current metadata.
    if let (Some(head), Some(snapshot)) = (new_head, minted)
        && snapshot.snapshot_id == head
    {
        view.snapshot = snapshot;
    }

    Ok(())
}

/// Removes exactly the entity at one `current` key, keeping the name indexes
/// coherent but cascading to nothing — the store deletes each ended entity
/// under its own key.
fn remove_entity(view: &mut CatalogSnapshot, entity: EntityKey) -> Result<()> {
    match entity {
        EntityKey::Schema { schema_id } => view.remove_schema_only(schema_id),
        EntityKey::Table { table_id } => view.remove_table_only(table_id),
        EntityKey::View { view_id } => view.delete_view(view_id),
        EntityKey::Column {
            table_id,
            column_id,
        } => view.remove_column_only(table_id, column_id),
        EntityKey::File {
            table_id,
            data_file_id,
        } => view.delete_data_file(table_id, data_file_id),
        EntityKey::DeleteFile {
            table_id,
            delete_file_id,
        } => view.delete_delete_file(table_id, delete_file_id),
        EntityKey::Partition {
            table_id,
            partition_id,
        } => view.delete_partition(table_id, partition_id),
        EntityKey::Sort { table_id, sort_id } => view.delete_sort(table_id, sort_id),
        EntityKey::Macro { macro_id } => view.delete_macro(macro_id),
        EntityKey::Index { table_id, index_id } => view.delete_index(table_id, index_id),
        EntityKey::FileColumnStats {
            table_id,
            data_file_id,
            column_id,
        } => view.remove_file_column_stats(table_id, data_file_id, column_id),
        EntityKey::TableStats { table_id } => view.remove_table_stats(table_id),
        EntityKey::TableColumnStats {
            table_id,
            column_id,
        } => view.remove_table_column_stats(table_id, column_id),
        EntityKey::Option {
            scope_kind,
            scope_id,
        } => view.remove_option_record((scope_kind, scope_id)),
        EntityKey::Tag { object_id } => view.remove_tag(object_id),
        EntityKey::Mapping { .. } => {
            return Err(Error::Corruption(
                "column-mapping deletion is unexpected; mappings are immutable".to_string(),
            ));
        }
    }
    Ok(())
}