timelake 0.7.0

The open time-series lakehouse: fast queries on your own object storage, with no database and no maintenance jobs
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

//! Concurrent commits: writers that lose the race rebase or report a conflict, and leave no
//! files behind.

#![allow(clippy::unwrap_used, reason = "test code")]

mod common;

use std::path::{Path, PathBuf};

use common::{T0, batch, opts, scalar_i64, schema, temp_lake};
use timelake::{AppendOptions, Error, MaintainOptions, Session};

fn no_maintenance() -> AppendOptions {
    AppendOptions { maintain: Some(false), ..AppendOptions::default() }
}

fn compact_only() -> MaintainOptions {
    MaintainOptions { retention: false, expire: false, ..MaintainOptions::default() }
}

/// Every file under `dir`, recursively.
fn files_under(dir: &Path) -> Vec<PathBuf> {
    let mut out = vec![];
    for entry in std::fs::read_dir(dir).unwrap() {
        let path = entry.unwrap().path();
        if path.is_dir() { out.extend(files_under(&path)) } else { out.push(path) }
    }
    out
}

/// How many files under `dir` have a name containing `pattern`.
fn count(dir: &Path, pattern: &str) -> usize {
    files_under(dir).iter().filter(|p| p.file_name().unwrap().to_str().unwrap().contains(pattern)).count()
}

/// Compaction planned against an old version still commits after an append: the files it
/// replaces are all still there.
#[tokio::test]
async fn compaction_rebases_over_a_concurrent_append() {
    let (_dir, lake) = temp_lake();
    let t = lake.create_table("iot.r", &schema(), opts()).await.unwrap();
    let (t, _) = t.append(&[batch(T0, 1, 1, 2, |_, _| 1.0)], no_maintenance()).await.unwrap();
    let (stale, _) = t.append(&[batch(T0 + 1, 1, 1, 2, |_, _| 2.0)], no_maintenance()).await.unwrap();
    let (latest, _) = stale.append(&[batch(T0 + 2, 1, 1, 2, |_, _| 3.0)], no_maintenance()).await.unwrap();

    // Compact the first two files from the stale version: rebases over the third append.
    let (after, report) = stale.maintain(compact_only()).await.unwrap();
    assert_eq!(report.files_rewritten, 2);
    assert_eq!(after.version(), latest.version() + 1);
    assert_eq!(after.files().await.unwrap().len(), 2);
    assert_eq!(scalar_i64(&Session::new(lake), "SELECT count(*) FROM iot.r").await, 6);
}

/// Two compactions of the same files: the second finds them gone, reports a conflict, and
/// deletes what it wrote.
#[tokio::test]
async fn compactions_of_the_same_files_conflict() {
    let (dir, lake) = temp_lake();
    let t = lake.create_table("iot.r", &schema(), opts()).await.unwrap();
    let (t, _) = t.append(&[batch(T0, 1, 1, 2, |_, _| 1.0)], no_maintenance()).await.unwrap();
    let (t, _) = t.append(&[batch(T0 + 1, 1, 1, 2, |_, _| 2.0)], no_maintenance()).await.unwrap();

    let (first, _) = t.maintain(compact_only()).await.unwrap();
    let err = t.maintain(compact_only()).await.unwrap_err();
    assert!(matches!(err, Error::Conflict(_)), "{err}");

    // The two original files (not expired yet) and the first compaction's output; nothing from
    // the second.
    assert_eq!(count(&dir.path().join("iot/r/data"), ".parquet"), 3);
    assert_eq!(first.files().await.unwrap().len(), 1);
    assert_eq!(scalar_i64(&Session::new(lake), "SELECT count(*) FROM iot.r").await, 4);
}

/// A writer that loses the race to a commit with the same batch id finds it on retry and
/// writes nothing.
#[tokio::test]
async fn batch_id_is_checked_again_after_a_lost_race() {
    let (dir, lake) = temp_lake();
    let base = lake.create_table("iot.r", &schema(), opts()).await.unwrap();
    let rows = batch(T0, 1, 1, 3, |_, _| 1.0);
    let id = AppendOptions { batch_id: Some("b1".into()), ..no_maintenance() };

    let (winner, _) = base.append(std::slice::from_ref(&rows), id.clone()).await.unwrap();
    let (loser, r) = base.append(&[rows], id).await.unwrap();
    assert!(r.commit.duplicate);
    assert_eq!((r.commit.attempts, r.rows), (2, 0));
    assert_eq!(loser.version(), winner.version());

    assert_eq!(count(&dir.path().join("iot/r/data"), ".parquet"), 1);
    assert_eq!(count(&dir.path().join("iot/r/metadata"), "lvi-"), 1);
    assert_eq!(scalar_i64(&Session::new(lake), "SELECT count(*) FROM iot.r").await, 3);
}

/// The losing attempt's manifests, manifest list and last-value index are deleted before it
/// rebases; only the files of the two committed snapshots remain.
#[tokio::test]
async fn lost_race_leaves_no_files() {
    let (dir, lake) = temp_lake();
    let base = lake.create_table("iot.r", &schema(), opts()).await.unwrap();
    let (_, first) = base.append(&[batch(T0, 1, 1, 3, |_, _| 1.0)], no_maintenance()).await.unwrap();
    let (t, second) = base.append(&[batch(T0 + 1, 1, 1, 3, |_, _| 2.0)], no_maintenance()).await.unwrap();
    assert_eq!((first.commit.attempts, second.commit.attempts), (1, 2));
    assert_eq!(t.metadata().snapshots().len(), 2);

    let metadata = dir.path().join("iot/r/metadata");
    assert_eq!(count(&metadata, "-m.avro"), 2);
    assert_eq!(count(&metadata, "snap-"), 2);
    assert_eq!(count(&metadata, "lvi-"), 2);
    assert_eq!(count(&dir.path().join("iot/r/data"), ".parquet"), 2);
    assert_eq!(scalar_i64(&Session::new(lake), "SELECT count(*) FROM iot.r").await, 6);
}