nodedb 0.4.0

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
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

// SPDX-License-Identifier: BUSL-1.1

use super::*;
use nodedb_types::Value;
use nodedb_types::protocol::opcodes::ResponseStatus;
use nodedb_types::protocol::{MAX_FRAME_SIZE, NativeResponse};

#[test]
fn chunk_large_response_splits_rows() {
    // Build a response with 100 rows, each ~200 bytes when serialized.
    let columns = vec!["id".to_string(), "data".to_string()];
    let rows: Vec<Vec<Value>> = (0..100)
        .map(|i| {
            vec![
                Value::Integer(i),
                Value::String(format!("row-data-{i}-padding-{}", "x".repeat(150))),
            ]
        })
        .collect();

    let response = NativeResponse {
        seq: 1,
        status: ResponseStatus::Ok,
        columns: Some(columns),
        rows: Some(rows),
        rows_affected: None,
        watermark_lsn: 42,
        error: None,
        auth: None,
        warnings: Vec::new(),
    };

    let frames = chunk_large_response(response, codec::FrameFormat::MessagePack).unwrap();

    // With 100 rows of ~200 bytes each (~20KB total), this should fit in
    // one frame (MAX_FRAME_SIZE = 16MB). Test with a scenario that forces splitting.
    assert!(!frames.is_empty());

    // Decode each frame and verify structure.
    for (i, frame) in frames.iter().enumerate() {
        let resp: NativeResponse = zerompk::from_msgpack(frame).unwrap();
        assert!(resp.rows.is_some());
        if i < frames.len() - 1 {
            assert_eq!(resp.status, ResponseStatus::Partial);
        } else {
            assert_eq!(resp.status, ResponseStatus::Ok);
        }
    }
}

#[test]
fn chunk_large_response_no_rows_passthrough() {
    let response = NativeResponse {
        seq: 1,
        status: ResponseStatus::Ok,
        columns: None,
        rows: None,
        rows_affected: Some(5),
        watermark_lsn: 42,
        error: None,
        auth: None,
        warnings: Vec::new(),
    };

    let frames = chunk_large_response(response, codec::FrameFormat::MessagePack).unwrap();
    assert_eq!(
        frames.len(),
        1,
        "no-rows response should pass through as-is"
    );
}

#[test]
fn chunk_large_response_preserves_all_rows() {
    // Create a response that's guaranteed to exceed MAX_FRAME_SIZE.
    // Each row ~200 bytes * 100K rows = ~20MB > 16MB limit.
    let columns = vec!["id".to_string(), "value".to_string()];
    let row_count = 100_000;
    let rows: Vec<Vec<Value>> = (0..row_count)
        .map(|i| {
            vec![
                Value::Integer(i),
                Value::String(format!("v{i}-{}", "p".repeat(150))),
            ]
        })
        .collect();

    let response = NativeResponse {
        seq: 42,
        status: ResponseStatus::Ok,
        columns: Some(columns.clone()),
        rows: Some(rows),
        rows_affected: None,
        watermark_lsn: 99,
        error: None,
        auth: None,
        warnings: Vec::new(),
    };

    let frames = chunk_large_response(response, codec::FrameFormat::MessagePack).unwrap();
    assert!(frames.len() > 1, "should produce multiple frames");

    // Reassemble all rows from frames (simulating client behavior).
    let mut total_rows: Vec<Vec<Value>> = Vec::new();
    for frame in &frames {
        let resp: NativeResponse = zerompk::from_msgpack(frame).unwrap();
        if let Some(rows) = resp.rows {
            total_rows.extend(rows);
        }
    }
    assert_eq!(total_rows.len(), row_count as usize);

    // First frame should have columns.
    let first: NativeResponse = zerompk::from_msgpack(&frames[0]).unwrap();
    assert_eq!(first.columns, Some(columns));
    assert_eq!(first.status, ResponseStatus::Partial);

    // Last frame should have Ok status.
    let last: NativeResponse = zerompk::from_msgpack(frames.last().unwrap()).unwrap();
    assert_eq!(last.status, ResponseStatus::Ok);

    // Each frame should be <= MAX_FRAME_SIZE.
    for frame in &frames {
        assert!(
            frame.len() <= MAX_FRAME_SIZE as usize,
            "frame size {} exceeds MAX_FRAME_SIZE {}",
            frame.len(),
            MAX_FRAME_SIZE
        );
    }
}