deltalake 0.3.0

Native Delta Lake implementation in Rust
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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148

extern crate chrono;
extern crate deltalake;
extern crate utime;

#[cfg(feature = "s3")]
#[allow(dead_code)]
mod s3_common;

#[allow(dead_code)]
mod fs_common;

use std::collections::HashMap;

use deltalake::{action, DeltaTransactionError};

#[tokio::test]
async fn test_two_commits_fs() {
    prepare_fs();
    test_two_commits("./tests/data/simple_commit")
        .await
        .unwrap();
}

#[cfg(all(feature = "s3", feature = "dynamodb"))]
#[tokio::test]
async fn test_two_commits_s3() {
    let path = "s3://deltars/simple_commit_rw1";
    s3_common::setup_dynamodb("concurrent_writes");
    prepare_s3(path).await;

    test_two_commits(path).await.unwrap();
}

#[cfg(all(feature = "s3", not(feature = "dynamodb")))]
#[tokio::test]
async fn test_two_commits_s3_fails_with_no_lock() {
    use deltalake::{StorageError, TransactionCommitAttemptError};

    let path = "s3://deltars/simple_commit_rw2";
    prepare_s3(path).await;

    let result = test_two_commits(path).await;
    if let Err(DeltaTransactionError::CommitRetriesExceeded { ref inner }) = result {
        if let TransactionCommitAttemptError::Storage { source } = inner {
            if let StorageError::S3Generic(err) = source {
                assert_eq!(err, "dynamodb locking is not enabled");
                return;
            }
        }
    }

    result.unwrap();

    panic!("S3 commit without dynamodb locking is expected to fail")
}

async fn test_two_commits(table_path: &str) -> Result<(), DeltaTransactionError> {
    let mut table = deltalake::open_table(table_path).await?;

    assert_eq!(0, table.version);
    assert_eq!(0, table.get_files().len());

    let tx1_actions = vec![
        action::Action::add(action::Add {
            path: String::from(
                "part-00000-b44fcdb0-8b06-4f3a-8606-f8311a96f6dc-c000.snappy.parquet",
            ),
            size: 396,
            partitionValues: HashMap::new(),
            partitionValues_parsed: None,
            modificationTime: 1564524294000,
            dataChange: true,
            stats: None,
            stats_parsed: None,
            tags: None,
        }),
        action::Action::add(action::Add {
            path: String::from(
                "part-00001-185eca06-e017-4dea-ae49-fc48b973e37e-c000.snappy.parquet",
            ),
            size: 400,
            partitionValues: HashMap::new(),
            partitionValues_parsed: None,
            modificationTime: 1564524294000,
            dataChange: true,
            stats: None,
            stats_parsed: None,
            tags: None,
        }),
    ];

    let mut tx1 = table.create_transaction(None);
    let version = tx1.commit_with(tx1_actions.as_slice(), None).await?;

    assert_eq!(1, version);
    assert_eq!(version, table.version);
    assert_eq!(2, table.get_files().len());

    let tx2_actions = vec![
        action::Action::add(action::Add {
            path: String::from(
                "part-00000-512e1537-8aaa-4193-b8b4-bef3de0de409-c000.snappy.parquet",
            ),
            size: 396,
            partitionValues: HashMap::new(),
            partitionValues_parsed: None,
            modificationTime: 1564524296000,
            dataChange: true,
            stats: None,
            stats_parsed: None,
            tags: None,
        }),
        action::Action::add(action::Add {
            path: String::from(
                "part-00001-4327c977-2734-4477-9507-7ccf67924649-c000.snappy.parquet",
            ),
            size: 400,
            partitionValues: HashMap::new(),
            partitionValues_parsed: None,
            modificationTime: 1564524296000,
            dataChange: true,
            stats: None,
            stats_parsed: None,
            tags: None,
        }),
    ];

    let mut tx2 = table.create_transaction(None);
    let version = tx2.commit_with(tx2_actions.as_slice(), None).await.unwrap();

    assert_eq!(2, version);
    assert_eq!(version, table.version);
    assert_eq!(4, table.get_files().len());
    Ok(())
}

fn prepare_fs() {
    fs_common::cleanup_dir_except(
        "./tests/data/simple_commit/_delta_log",
        vec!["00000000000000000000.json".to_string()],
    );
}

#[cfg(feature = "s3")]
async fn prepare_s3(path: &str) {
    let delta_log = format!("{}/_delta_log", path);
    s3_common::cleanup_dir_except(&delta_log, vec!["00000000000000000000.json".to_string()]).await;
}