moraine 0.6.5

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

//! Inline source traversal with one decoded chunk retained.

use arrow::datatypes::SchemaRef;

use super::{ColumnId, Error, InlineBuildCursorValue, Result, StepBuffer, backfill};
use crate::{
    catalog::TableId,
    data_file,
    store::{
        handle::ReadHandle,
        inline::{
            self as store_inline,
            stream::{InlineChunks, InlineTombstones},
        },
        key::InlineOperation,
    },
};

pub(super) async fn stream_inline_sources(
    source: backfill::BackfillSource<'_>,
    columns: &[ColumnId],
    legacy_cursor: Option<u64>,
    buffer: &mut StepBuffer<'_>,
) -> Result<()> {
    let backfill::BackfillSource {
        snapshot,
        table,
        handle,
    } = source;
    let positions = snapshot.column_positions(table, columns)?;
    let initial_cursor = buffer.inline_cursor;
    let mut chunks = InlineChunks::open(handle, table.get(), initial_cursor.as_ref()).await?;
    let mut schema = None;
    while let Some((operation, chunk)) = chunks.next().await? {
        let InlineOperation::Insert {
            schema_version,
            begin_snapshot,
            chunk_seq,
            ..
        } = operation
        else {
            return Err(Error::Corruption("non-insert inline build source".into()));
        };
        let current_schema = cached_schema(handle, table, schema_version, &mut schema).await?;
        let start = initial_cursor
            .as_ref()
            .filter(|cursor| {
                (
                    cursor.schema_version,
                    cursor.begin_snapshot,
                    cursor.chunk_seq,
                ) == (schema_version, begin_snapshot, chunk_seq)
            })
            .map_or(0, |cursor| cursor.next_position);
        let columns = snapshot
            .inline_read_columns(handle, table, begin_snapshot)
            .await?;
        let mut rows = data_file::InlineRows::new(
            current_schema,
            &chunk.body,
            &positions,
            chunk.row_id_start,
            start,
            &columns,
        )?;
        if u64::try_from(rows.row_count()).ok() != Some(chunk.row_count) {
            return Err(Error::Corruption(
                "inline chunk row count disagrees with its body".into(),
            ));
        }
        buffer.peak_inline_body_bytes = buffer.peak_inline_body_bytes.max(chunk.body.len());
        buffer.peak_inline_decoded_bytes =
            buffer.peak_inline_decoded_bytes.max(rows.decoded_bytes());
        let end = chunk
            .row_id_start
            .checked_add(chunk.row_count.saturating_sub(1))
            .ok_or_else(|| Error::Corruption("inline chunk row range overflow".into()))?;
        let mut tombstones =
            InlineTombstones::open(handle, table.get(), chunk.row_id_start, end).await?;
        while let Some(entry) = rows.next()? {
            let next_position = entry.ordinal + 1;
            let dead = tombstones
                .latest(entry.row_id)
                .await?
                .is_some_and(|end| begin_snapshot < end);
            if !dead && legacy_cursor.is_none_or(|cursor| entry.row_id > cursor) {
                buffer.push_values(entry.row_id, &entry.values).await?;
            }
            buffer.inline_cursor = Some(InlineBuildCursorValue {
                schema_version,
                begin_snapshot,
                chunk_seq,
                next_position,
                chunk_finished: next_position == chunk.row_count,
                complete: false,
                covered_snapshot: 0,
            });
        }
    }
    Ok(())
}

async fn cached_schema(
    handle: ReadHandle<'_>,
    table: TableId,
    version: u64,
    cached: &mut Option<(u64, SchemaRef)>,
) -> Result<SchemaRef> {
    if let Some((current, schema)) = cached
        && *current == version
    {
        return Ok(schema.clone());
    }
    let bytes = store_inline::read_inline_schema(handle, table.get(), version)
        .await?
        .ok_or_else(|| {
            Error::Corruption(format!(
                "no inline schema for table {table} version {version}"
            ))
        })?;
    let schema = data_file::decode_inline_schema(bytes)?;
    *cached = Some((version, schema.clone()));
    Ok(schema)
}