jetro-core 0.5.12

jetro-core: parser, compiler, and VM for the Jetro JSON query language
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

use super::RowError;
use memchr::memchr;
use std::ops::Range;

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum NdjsonRowFrame {
    JsonLine,
    DelimitedPayload {
        separator: u8,
        null_payload: NullPayload,
    },
}

impl Default for NdjsonRowFrame {
    fn default() -> Self {
        Self::JsonLine
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum NullPayload {
    Skip,
    Keep,
    Error,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub(super) enum FramePayload {
    Data(Range<usize>),
    Skip,
}

#[inline]
pub(super) fn frame_payload(
    frame: NdjsonRowFrame,
    line_no: u64,
    row: &[u8],
) -> Result<FramePayload, RowError> {
    let range = match frame {
        NdjsonRowFrame::JsonLine => 0..row.len(),
        NdjsonRowFrame::DelimitedPayload { separator, .. } => {
            let Some(sep) = memchr(separator, row) else {
                return Ok(FramePayload::Skip);
            };
            sep + 1..row.len()
        }
    };
    let range = trim_range(row, range);
    if range.is_empty() {
        return match frame {
            NdjsonRowFrame::JsonLine => Err(RowError::EmptyPayload { line_no }),
            NdjsonRowFrame::DelimitedPayload { .. } => Ok(FramePayload::Skip),
        };
    }

    if let NdjsonRowFrame::DelimitedPayload { null_payload, .. } = frame {
        if &row[range.clone()] == b"null" {
            return match null_payload {
                NullPayload::Skip => Ok(FramePayload::Skip),
                NullPayload::Keep => Ok(FramePayload::Data(range)),
                NullPayload::Error => Err(RowError::NullPayload { line_no }),
            };
        }
        if !payload_starts_like_json(&row[range.clone()]) {
            return Ok(FramePayload::Skip);
        }
    }

    Ok(FramePayload::Data(range))
}

#[inline]
fn payload_starts_like_json(payload: &[u8]) -> bool {
    matches!(
        payload[0],
        b'{' | b'[' | b'"' | b't' | b'f' | b'-' | b'0'..=b'9'
    )
}

#[inline]
fn trim_range(row: &[u8], range: Range<usize>) -> Range<usize> {
    let mut start = range.start;
    let mut end = range.end;
    while start < end && row[start].is_ascii_whitespace() {
        start += 1;
    }
    while end > start && row[end - 1].is_ascii_whitespace() {
        end -= 1;
    }
    start..end
}

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

    #[test]
    fn delimited_payload_skips_null() {
        let frame = NdjsonRowFrame::DelimitedPayload {
            separator: b'|',
            null_payload: NullPayload::Skip,
        };

        assert_eq!(
            frame_payload(frame, 1, b"k|null").unwrap(),
            FramePayload::Skip
        );
        assert_eq!(
            frame_payload(frame, 2, br#"k| {"id":1} "#).unwrap(),
            FramePayload::Data(3..11)
        );
        assert_eq!(frame_payload(frame, 3, b"k|").unwrap(), FramePayload::Skip);
        assert_eq!(
            frame_payload(frame, 4, b"no-separator").unwrap(),
            FramePayload::Skip
        );
        assert_eq!(
            frame_payload(frame, 5, b"k|not-json").unwrap(),
            FramePayload::Skip
        );
    }
}