dbn-cli 0.56.0

Command-line utility for converting Databento Binary Encoding (DBN) files to text-based formats
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

use std::num::NonZeroU64;

use dbn::{
    decode::{DbnMetadata, DecodeRecordRef},
    RType, Record, RecordRef, Schema,
};

#[derive(Debug)]
pub struct SchemaFilter<D> {
    decoder: D,
    rtype: Option<RType>,
}

impl<D> SchemaFilter<D>
where
    D: DbnMetadata,
{
    pub fn new(mut decoder: D, schema: Option<Schema>) -> Self {
        if let Some(schema) = schema {
            decoder.metadata_mut().schema = Some(schema);
        }
        Self::new_no_metadata(decoder, schema)
    }
}

impl<D> SchemaFilter<D> {
    pub fn new_no_metadata(decoder: D, schema: Option<Schema>) -> Self {
        Self {
            decoder,
            rtype: schema.map(RType::from),
        }
    }
}

impl<D: DbnMetadata> DbnMetadata for SchemaFilter<D> {
    fn metadata(&self) -> &dbn::Metadata {
        self.decoder.metadata()
    }

    fn metadata_mut(&mut self) -> &mut dbn::Metadata {
        self.decoder.metadata_mut()
    }
}

impl<D: DecodeRecordRef> DecodeRecordRef for SchemaFilter<D> {
    fn decode_record_ref(&mut self) -> dbn::Result<Option<dbn::RecordRef<'_>>> {
        while let Some(record) = self.decoder.decode_record_ref()? {
            if self
                .rtype
                .is_none_or(|rtype| rtype as u8 == record.header().rtype)
            {
                // Safe: casting reference to pointer so the pointer will always be valid.
                // Getting around borrow checker limitation.
                return Ok(Some(unsafe {
                    RecordRef::unchecked_from_header(record.header())
                }));
            }
        }
        Ok(None)
    }
}

#[derive(Debug)]
pub struct LimitFilter<D> {
    decoder: D,
    limit: Option<NonZeroU64>,
    record_count: u64,
}

impl<D> LimitFilter<D>
where
    D: DbnMetadata,
{
    pub fn new(mut decoder: D, limit: Option<NonZeroU64>) -> Self {
        if let Some(limit) = limit {
            let metadata_limit = &mut decoder.metadata_mut().limit;
            if let Some(metadata_limit) = metadata_limit {
                *metadata_limit = (*metadata_limit).min(limit);
            } else {
                *metadata_limit = Some(limit);
            }
        }
        Self::new_no_metadata(decoder, limit)
    }
}

impl<D> LimitFilter<D> {
    pub fn new_no_metadata(decoder: D, limit: Option<NonZeroU64>) -> Self {
        Self {
            decoder,
            limit,
            record_count: 0,
        }
    }
}

impl<D: DbnMetadata> DbnMetadata for LimitFilter<D> {
    fn metadata(&self) -> &dbn::Metadata {
        self.decoder.metadata()
    }

    fn metadata_mut(&mut self) -> &mut dbn::Metadata {
        self.decoder.metadata_mut()
    }
}

impl<D: DecodeRecordRef> DecodeRecordRef for LimitFilter<D> {
    fn decode_record_ref(&mut self) -> dbn::Result<Option<RecordRef<'_>>> {
        if self
            .limit
            .is_some_and(|limit| self.record_count >= limit.get())
        {
            return Ok(None);
        }
        Ok(self.decoder.decode_record_ref()?.inspect(|_| {
            self.record_count += 1;
        }))
    }
}