bkndb 0.2.3

Embedded hybrid graph-relational database engine with ACID transactions
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

use bkndb::graph::NodeId;
use bkndb::value::{PropValue, Properties};
use bkndb::{BknDb, SyncBatch};
use std::time::{Duration, Instant};

#[test]
fn test_50k_nodes_and_50k_edges_bulk_sync_benchmark() {
    let dir = tempfile::tempdir().unwrap();
    let db_path = dir.path().join("bench_50k.bkndb");

    let db = BknDb::open(&db_path).expect("failed to open on-disk database");

    const COUNT: usize = 50_000;

    println!(
        "Generating {} nodes and {} edges in memory...",
        COUNT, COUNT
    );
    let mut batch = SyncBatch::new();
    batch.nodes.reserve(COUNT);
    batch.edges.reserve(COUNT);

    for i in 1..=COUNT {
        let mut props = Properties::new();
        props.insert("idx".to_string(), PropValue::Int(i as i64));
        batch.nodes.push(("Symbol".to_string(), props));
    }

    // Connect node i to node i+1 (and wrap around at the end)
    for i in 1..=COUNT {
        let from = NodeId(i as u64);
        let to = if i == COUNT {
            NodeId(1)
        } else {
            NodeId((i + 1) as u64)
        };
        batch
            .edges
            .push((from, "calls".to_string(), to, Properties::new()));
    }

    println!("Starting bulk atomic ingestion of 50k nodes + 50k edges...");
    let start = Instant::now();
    let res = db.sync_batch(batch).expect("bulk sync_batch failed");
    let elapsed = start.elapsed();

    println!(
        "Ingested {} nodes and {} edges in {:?} ({:.2} ops/sec)",
        res.node_ids.len(),
        res.edge_ids.len(),
        elapsed,
        (COUNT * 2) as f64 / elapsed.as_secs_f64()
    );

    assert_eq!(res.node_ids.len(), COUNT);
    assert_eq!(res.edge_ids.len(), COUNT);

    // In debug mode (unoptimized), allow up to 25s for parallel test runs; in release mode (optimized), target < 5000ms to tolerate shared CI disk I/O.
    let max_duration = if cfg!(debug_assertions) {
        Duration::from_secs(25)
    } else {
        Duration::from_millis(5000)
    };
    assert!(
        elapsed < max_duration,
        "Ingestion took {:?}, exceeding maximum allowable duration {:?}",
        elapsed,
        max_duration
    );

    // Verification: sample random nodes and verify properties and neighbors
    db.read_tx(|tx| {
        for &sample_id in &[1, 25_000, 50_000] {
            let nid = NodeId(sample_id);
            let node = tx.graph().get_node(nid)?.expect("sampled node must exist");
            assert_eq!(node.label, "Symbol");
            assert_eq!(
                node.properties.get("idx"),
                Some(&PropValue::Int(sample_id as i64))
            );

            let neighbors = tx.graph().neighbors_out(nid, "calls")?;
            assert_eq!(neighbors.len(), 1);
            let expected_to = if sample_id == COUNT as u64 {
                NodeId(1)
            } else {
                NodeId(sample_id + 1)
            };
            assert_eq!(neighbors[0].0, expected_to);
        }
        Ok::<_, bkndb::BknError>(())
    })
    .unwrap();
}