faucet-source-dynamodb 1.2.1

Amazon DynamoDB source (parallel scan, query, Streams CDC) for the faucet-stream ecosystem
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

//! The connector's static descriptor (RFC 0013).

use faucet_core::capabilities::config::str_field;
use faucet_core::{Capabilities, ConnectorDescriptor, Value};

/// The connector's static descriptor (RFC 0013): its capabilities, answered from config alone.
pub const DESCRIPTOR: ConnectorDescriptor = ConnectorDescriptor::new("dynamodb", capabilities);

fn capabilities(config: &Value) -> Capabilities {
    let builder = Capabilities::builder().discover(true);
    if config.is_null() {
        return builder
            .shardable(true)
            .reports_lag(true)
            .table_routing(true)
            .complete_reads(true)
            .build();
    }
    let mode = str_field(config, "mode").unwrap_or("scan");
    let resnapshot = mode == "streams"
        && str_field(config, "on_gap") == Some("resnapshot")
        && config.get("resnapshot_deletes") != Some(&Value::Bool(false));
    builder
        .shardable(!matches!(mode, "query" | "streams"))
        .reports_lag(mode == "streams")
        .table_routing(mode == "streams")
        .complete_reads(resnapshot)
        .build()
}

#[cfg(test)]
mod tests {
    use super::*;
    use faucet_core::json;

    #[test]
    fn mode_decides_sharding_and_lag() {
        let caps = |c: Value| DESCRIPTOR.capabilities(&c);
        for c in [
            json!({}),
            json!({"mode": "scan"}),
            json!({"mode": "${env:M}"}),
        ] {
            let s = caps(c);
            assert!(s.discover() && s.shardable() && !s.reports_lag());
        }
        let q = caps(json!({"mode": "query"}));
        assert!(q.discover() && !q.shardable() && !q.reports_lag());
        let st = caps(json!({"mode": "streams"}));
        assert!(st.discover() && !st.shardable() && st.reports_lag());
        assert!(st.table_routing() && !q.table_routing());
        assert!(caps(Value::Null).table_routing());
        assert!(!st.complete_reads() && caps(Value::Null).complete_reads());
        assert!(caps(json!({"mode": "streams", "on_gap": "resnapshot"})).complete_reads());
        assert!(!caps(json!({"mode": "scan", "on_gap": "resnapshot"})).complete_reads());
        assert!(
            !caps(json!({"mode": "streams", "on_gap": "resnapshot", "resnapshot_deletes": false}))
                .complete_reads()
        );
        assert!(
            caps(json!({"mode": "streams", "on_gap": "resnapshot", "resnapshot_deletes": true}))
                .complete_reads()
        );
        let kind = caps(Value::Null);
        assert!(kind.discover() && kind.shardable() && kind.reports_lag());
        assert!(!kind.exactly_once() && !kind.columnar() && !kind.native());
        assert_eq!(DESCRIPTOR.kind(), "dynamodb");
    }
}