datui_lib/formats/
layout.rs

1//! A file's header, footer and lookups read, and where each record field sits.
2
3use super::*;
4
5/// A header, read: each named field's value, and how many bytes the header takes;
6/// and the footer's values, and the symbol lists fields index into.
7#[derive(Debug, Default, Clone)]
8pub struct HeaderValues {
9    pub values: Vec<(String, AnyValue<'static>)>,
10    pub size: u64,
11    pub footer: Vec<(String, AnyValue<'static>)>,
12    pub lookups: BTreeMap<String, Arc<Vec<String>>>,
13}
14
15fn int_value(value: &AnyValue<'_>) -> Option<i128> {
16    match value {
17        AnyValue::UInt8(v) => Some(i128::from(*v)),
18        AnyValue::UInt16(v) => Some(i128::from(*v)),
19        AnyValue::UInt32(v) => Some(i128::from(*v)),
20        AnyValue::UInt64(v) => Some(i128::from(*v)),
21        AnyValue::Int8(v) => Some(i128::from(*v)),
22        AnyValue::Int16(v) => Some(i128::from(*v)),
23        AnyValue::Int32(v) => Some(i128::from(*v)),
24        AnyValue::Int64(v) => Some(i128::from(*v)),
25        _ => None,
26    }
27}
28
29impl HeaderValues {
30    fn get(&self, name: &str) -> Option<&AnyValue<'static>> {
31        self.values.iter().find(|(n, _)| n == name).map(|(_, v)| v)
32    }
33
34    fn footer_int(&self, name: &str) -> Option<i128> {
35        self.footer
36            .iter()
37            .find(|(n, _)| n == name)
38            .and_then(|(_, v)| int_value(v))
39    }
40
41    /// `amount`'s value with no bound but its sign: an offset or a count, which the
42    /// file's length bounds where it is used.
43    pub(crate) fn resolve_any(&self, amount: &Amount, what: &str) -> Result<u64, String> {
44        let (value, field) = match amount {
45            Amount::Given(n) => return Ok(*n),
46            Amount::Header { field, adjust } => (
47                self.int(field).map(|v| v + i128::from(*adjust)),
48                format!("header's `{field}`"),
49            ),
50            Amount::Footer { field, adjust } => (
51                self.footer_int(field).map(|v| v + i128::from(*adjust)),
52                format!("footer's `{field}`"),
53            ),
54            Amount::Record { .. } | Amount::Rest => {
55                return Err(format!("{what}: comes from each record"));
56            }
57        };
58        let value = value.ok_or_else(|| format!("{what}: the {field} has no value"))?;
59        u64::try_from(value).map_err(|_| format!("{what}: the {field} gives {value}, below 0"))
60    }
61
62    pub(crate) fn int(&self, name: &str) -> Option<i128> {
63        int_value(self.get(name)?)
64    }
65
66    pub(crate) fn text(&self, name: &str) -> Option<String> {
67        match self.get(name)? {
68            AnyValue::String(s) => Some(s.to_string()),
69            AnyValue::StringOwned(s) => Some(s.to_string()),
70            _ => None,
71        }
72    }
73
74    /// Nanoseconds from the Unix epoch to midnight of the date header field `name`
75    /// holds: a date, a datetime, or text such as 2024-01-02 or 20240102.
76    fn midnight_ns(&self, name: &str) -> Option<i64> {
77        let days = match self.get(name)? {
78            AnyValue::Date(days) => i64::from(*days),
79            AnyValue::Datetime(v, unit, _) | AnyValue::DatetimeOwned(v, unit, _) => {
80                let per_day = DAY_NS
81                    / match unit {
82                        TimeUnit::Nanoseconds => 1,
83                        TimeUnit::Microseconds => 1_000,
84                        TimeUnit::Milliseconds => 1_000_000,
85                    };
86                v.div_euclid(per_day)
87            }
88            other => {
89                let text = match other {
90                    AnyValue::String(s) => s.to_string(),
91                    AnyValue::StringOwned(s) => s.to_string(),
92                    _ => return None,
93                };
94                let date = chrono::NaiveDate::parse_from_str(&text, "%Y-%m-%d")
95                    .or_else(|_| chrono::NaiveDate::parse_from_str(&text, "%Y%m%d"))
96                    .ok()?;
97                (date - chrono::NaiveDate::from_ymd_opt(1970, 1, 1)?).num_days()
98            }
99        };
100        days.checked_mul(DAY_NS)
101    }
102
103    /// `amount`'s value: given, or read from a header or footer field and bounded.
104    pub(crate) fn resolve(&self, amount: &Amount, what: &str) -> Result<u64, String> {
105        match amount {
106            Amount::Given(n) => Ok(*n),
107            Amount::Record { .. } | Amount::Rest => Err(format!(
108                "{what}: comes from each record, which needs the records walked"
109            )),
110            Amount::Header { field, adjust } | Amount::Footer { field, adjust } => {
111                let (part, value) = match amount {
112                    Amount::Footer { .. } => ("footer", self.footer_int(field)),
113                    _ => ("header", self.int(field)),
114                };
115                let value = value
116                    .ok_or_else(|| format!("{what}: the {part} has no value for `{field}`"))?
117                    + i128::from(*adjust);
118                // A record count is bounded by the file instead.
119                let bound = if what == "count" {
120                    i128::from(u64::MAX)
121                } else {
122                    i128::from(MAX_SIZE)
123                };
124                if !(0..=bound).contains(&value) {
125                    return Err(format!(
126                        "{what}: the header's `{field}` gives {value}, outside 0 to {bound}"
127                    ));
128                }
129                Ok(value as u64)
130            }
131        }
132    }
133}
134
135/// Where a column's cells are: the first at `start`, `stride` apart (a cell's own
136/// width when `None`), each `count` values of `width` bytes.
137#[derive(Debug, Clone, Copy)]
138pub(crate) struct Place {
139    pub start: usize,
140    pub stride: Option<usize>,
141    pub width: usize,
142    pub count: usize,
143}
144
145/// The decoder's view of `field`, named `name`, its cells at `place`.
146pub(crate) fn layout_of(
147    spec: &Spec,
148    field: &Field,
149    name: &str,
150    place: Place,
151    header: &HeaderValues,
152) -> Result<ColumnLayout, String> {
153    let Place {
154        start,
155        stride,
156        width,
157        count,
158    } = place;
159    let logical = match &field.meaning {
160        Meaning::Plain if field.lookup.is_some() => Logical::Lookup(
161            field
162                .name
163                .as_ref()
164                .and_then(|n| header.lookups.get(n))
165                .cloned()
166                .ok_or_else(|| format!("{name}: its symbol list was not read"))?,
167        ),
168        Meaning::Plain => Logical::Plain,
169        Meaning::Scale(scale) => Logical::Decimal {
170            scale: *scale as usize,
171        },
172        Meaning::Linear { factor, offset } => Logical::Linear {
173            factor: *factor,
174            offset: *offset,
175        },
176        Meaning::Enum(labels) => Logical::Enum(labels.clone()),
177        Meaning::Yyyymmdd => Logical::Yyyymmdd,
178        Meaning::Time { unit, epoch_ns } => match (field.ty, unit) {
179            (Type::Float(_), unit) => Logical::FloatTimestamp {
180                ns_per_unit: unit.nanos() as f64,
181                epoch_ns: *epoch_ns,
182            },
183            (_, TimeUnitSpec::Days) => Logical::Days {
184                epoch_days: i32::try_from(epoch_ns.div_euclid(DAY_NS))
185                    .map_err(|_| format!("{name}: the epoch is out of range"))?,
186            },
187            (_, unit) => {
188                let (unit, multiplier, per) = match unit {
189                    TimeUnitSpec::Seconds => (TimeUnit::Milliseconds, 1000, 1_000_000),
190                    TimeUnitSpec::Millis => (TimeUnit::Milliseconds, 1, 1_000_000),
191                    TimeUnitSpec::Micros => (TimeUnit::Microseconds, 1, 1_000),
192                    _ => (TimeUnit::Nanoseconds, 1, 1),
193                };
194                Logical::Timestamp {
195                    unit,
196                    multiplier,
197                    epoch: epoch_ns / per,
198                }
199            }
200        },
201        Meaning::TimeOfDay { unit, date } => Logical::TimeOfDay {
202            ns_per_unit: unit.nanos(),
203            date_ns: match date {
204                None => None,
205                Some(field) => Some(
206                    header
207                        .midnight_ns(field)
208                        .ok_or_else(|| format!("{name}: the header's `{field}` holds no date"))?,
209                ),
210            },
211        },
212    };
213    Ok(ColumnLayout {
214        name: PlSmallStr::from(name),
215        source: 0,
216        start,
217        stride: stride.unwrap_or(width * count),
218        width,
219        count,
220        physical: if field.ty == Type::Str {
221            field.encoding.physical()
222        } else {
223            field.ty.physical()
224        },
225        big_endian: field.endian.unwrap_or(spec.endian) == Endian::Big,
226        null: field.null,
227        logical,
228    })
229}
230
231/// How many bytes one value of `field` takes and how many values it holds, now that
232/// the header is read.
233pub(crate) fn sized(field: &Field, header: &HeaderValues) -> Result<(u64, u64), String> {
234    let width = match (field.ty.width(), &field.size) {
235        (Some(w), _) => w,
236        (None, Some(amount)) => header.resolve(amount, "size")?,
237        (None, None) => 0,
238    };
239    let count = match &field.count {
240        None => 1,
241        Some(amount) => header.resolve(amount, "count")?,
242    };
243    if count > MAX_SIZE || width.saturating_mul(count) > MAX_SIZE {
244        return Err(format!(
245            "field `{}` would take {count} values of {width} bytes, more than {MAX_SIZE}",
246            field.name.as_deref().unwrap_or("pad")
247        ));
248    }
249    Ok((width, count))
250}
251
252/// Read `fields` from `bytes` at `at`, sizing each as it goes, into `read`'s header
253/// values or (for `footer`) its footer values. Gives where they end.
254fn read_fields(
255    spec: &Spec,
256    fields: &[Field],
257    bytes: &[u8],
258    mut at: u64,
259    read: &mut HeaderValues,
260    footer: bool,
261) -> Result<u64, String> {
262    for field in fields {
263        let (width, count) = sized(field, read)?;
264        let end = at + width * count;
265        if end > bytes.len() as u64 {
266            return Err(format!(
267                "the file is {} bytes, too short for its {} (field `{}` ends at byte {end})",
268                bytes.len(),
269                if footer { "footer" } else { "header" },
270                field.name.as_deref().unwrap_or("pad")
271            ));
272        }
273        if let Some(name) = &field.name
274            && width > 0
275        {
276            let place = Place {
277                start: at as usize,
278                stride: None,
279                width: width as usize,
280                count: count as usize,
281            };
282            let layout = layout_of(spec, field, name, place, read)?;
283            let column = crate::formats::fixed_records::decode(bytes, &layout, 1)
284                .map_err(|e| e.to_string())?;
285            let value = column.get(0).map_err(|e| e.to_string())?.into_static();
286            if footer {
287                read.footer.push((name.clone(), value));
288            } else {
289                read.values.push((name.clone(), value));
290            }
291        }
292        at = end;
293    }
294    Ok(at)
295}
296
297/// Read the header from the front of `bytes`, sizing each field as it goes.
298pub(crate) fn read_header(spec: &Spec, bytes: &[u8]) -> Result<HeaderValues, String> {
299    let mut read = HeaderValues::default();
300    let at = read_fields(spec, &spec.header.fields, bytes, 0, &mut read, false)?;
301    read.size = match &spec.header.size {
302        None => at,
303        Some(amount) => {
304            let size = read.resolve(amount, "header size")?;
305            if size < at {
306                return Err(format!(
307                    "the header's fields take {at} bytes, more than its size of {size}"
308                ));
309            }
310            size
311        }
312    };
313    Ok(read)
314}
315
316/// Read the footer at the end of `bytes` into `read`: where it starts, and a note on
317/// its checksum.
318pub(crate) fn read_footer(
319    spec: &Spec,
320    bytes: &[u8],
321    read: &mut HeaderValues,
322) -> Result<(u64, Option<String>), String> {
323    let len = bytes.len() as u64;
324    let Some(footer) = &spec.footer else {
325        return Ok((len, None));
326    };
327    let size = footer
328        .size
329        .or_else(|| given_width(&footer.fields))
330        .unwrap_or(0);
331    let start = len
332        .checked_sub(size)
333        .filter(|s| *s >= read.size)
334        .ok_or_else(|| {
335            format!(
336                "the file is {len} bytes, too short for its {}-byte header and {size}-byte footer",
337                read.size
338            )
339        })?;
340    read_fields(spec, &footer.fields, bytes, start, read, true)?;
341    // Notes are warnings: a checksum that matches says nothing.
342    let note = footer.checksum.as_ref().and_then(|(algo, field)| {
343        let stored = read.footer_int(field);
344        let computed = algo.compute(&bytes[..start as usize]);
345        match stored {
346            Some(v) if v == i128::from(computed) => None,
347            Some(v) => Some(format!(
348                "the footer's checksum `{field}` is {v:#x}; the file's is {computed:#x}"
349            )),
350            None => Some(format!(
351                "the footer has no value for its checksum `{field}`"
352            )),
353        }
354    });
355    Ok((start, note))
356}
357
358/// The symbol lists `fields` index into, read from beside `dir`.
359pub(crate) fn read_lookups(
360    fields: &[Field],
361    dir: Option<&Path>,
362    read: &mut HeaderValues,
363) -> Result<(), String> {
364    for field in fields {
365        let (Some(name), Some(lookup)) = (&field.name, &field.lookup) else {
366            continue;
367        };
368        let dir = dir.ok_or_else(|| {
369            format!("{name}: its symbol list {} is beside the data, which was not opened from a directory", lookup.file)
370        })?;
371        let path = dir.join(&lookup.file);
372        let size = std::fs::metadata(&path)
373            .map_err(|e| format!("{name}: the symbol list {}: {e}", path.display()))?
374            .len();
375        if size > MAX_SIZE {
376            return Err(format!(
377                "{name}: the symbol list {} is {size} bytes, more than {MAX_SIZE}",
378                path.display()
379            ));
380        }
381        let bytes = std::fs::read(&path)
382            .map_err(|e| format!("{name}: the symbol list {}: {e}", path.display()))?;
383        read.lookups
384            .insert(name.clone(), Arc::new(symbols(&bytes, lookup.format)));
385    }
386    Ok(())
387}
388
389/// The entries of a symbol list.
390pub fn symbols(bytes: &[u8], format: LookupFormat) -> Vec<String> {
391    match format {
392        LookupFormat::Lines => String::from_utf8_lossy(bytes)
393            .lines()
394            .map(|l| l.trim_end_matches('\r').to_string())
395            .collect(),
396        LookupFormat::Nul => {
397            let mut out: Vec<String> = bytes
398                .split(|b| *b == 0)
399                .map(|s| String::from_utf8_lossy(s).into_owned())
400                .collect();
401            // The list ends with a NUL, which leaves an empty piece after it.
402            if bytes.last() == Some(&0) {
403                out.pop();
404            }
405            out
406        }
407        LookupFormat::Fixed(n) => bytes
408            .chunks(n as usize)
409            .map(crate::formats::fixed_records::text)
410            .collect(),
411    }
412}
413
414/// The rows a spec reads, however they are framed: what the table, a window of it,
415/// `formats check` and the fuzz target read through.
416pub trait SpecRecords: crate::formats::pushdown::Windowed + std::fmt::Debug {
417    fn rows(&self) -> usize;
418    fn schema(&self) -> SchemaRef;
419    /// The frame: a scan that decodes only what a query asks for.
420    fn into_lazy(self: Arc<Self>) -> PolarsResult<LazyFrame>;
421    /// The first `rows` rows, decoded now.
422    fn collect(&self, rows: usize) -> PolarsResult<DataFrame>;
423    /// The bytes the rows are read from, for a reader of the raw bytes.
424    fn sources(&self) -> &[Arc<Bytes>];
425}
426
427impl std::fmt::Debug for FixedRecords {
428    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
429        f.debug_struct("FixedRecords")
430            .field("rows", &self.rows())
431            .finish()
432    }
433}
434
435impl SpecRecords for FixedRecords {
436    fn rows(&self) -> usize {
437        FixedRecords::rows(self)
438    }
439    fn schema(&self) -> SchemaRef {
440        FixedRecords::schema(self)
441    }
442    fn into_lazy(self: Arc<Self>) -> PolarsResult<LazyFrame> {
443        Ok(FixedRecords::lazy(&self))
444    }
445    fn collect(&self, rows: usize) -> PolarsResult<DataFrame> {
446        FixedRecords::collect(self, rows)
447    }
448    fn sources(&self) -> &[Arc<Bytes>] {
449        FixedRecords::sources(self)
450    }
451}
452
453impl SpecRecords for crate::formats::framed_records::FramedRecords {
454    fn rows(&self) -> usize {
455        crate::formats::framed_records::FramedRecords::rows(self)
456    }
457    fn schema(&self) -> SchemaRef {
458        crate::formats::framed_records::FramedRecords::schema(self)
459    }
460    fn into_lazy(self: Arc<Self>) -> PolarsResult<LazyFrame> {
461        Ok(crate::formats::framed_records::FramedRecords::lazy(&self))
462    }
463    fn collect(&self, rows: usize) -> PolarsResult<DataFrame> {
464        crate::formats::framed_records::FramedRecords::collect(self, rows)
465    }
466    fn sources(&self) -> &[Arc<Bytes>] {
467        crate::formats::framed_records::FramedRecords::sources(self)
468    }
469}
470
471/// A file read through a spec: its columns, and what the read had to say.
472pub struct Opened {
473    pub records: Arc<dyn SpecRecords>,
474    /// Warnings for the dataset's notes, one sentence each.
475    pub notes: Vec<String>,
476    pub header: HeaderValues,
477}
478
479/// A note for the records of `rows` past the most a table holds, which are not shown.
480pub(crate) fn past_limit(notes: &mut Vec<String>, rows: u64, records: &FixedRecords) {
481    let past = rows.saturating_sub(records.rows() as u64);
482    if past > 0 {
483        notes.push(format!(
484            "last {past} records not shown: past the table limit"
485        ));
486    }
487}
488
489/// Up to this many trailing bytes are shown in the warning about them.
490const TRAILING_SHOWN: usize = 32;
491
492pub(crate) fn trailing_note(what: &str, bytes: &[u8]) -> String {
493    let shown = &bytes[..bytes.len().min(TRAILING_SHOWN)];
494    let more = if bytes.len() > shown.len() {
495        " ..."
496    } else {
497        ""
498    };
499    format!(
500        "{what}: {} trailing {} left out, not a whole record: {}{more}",
501        bytes.len(),
502        if bytes.len() == 1 { "byte" } else { "bytes" },
503        crate::formats::fixed_records::hex(shown),
504    )
505}
datui_lib/formats/layout.rs

datui_lib/formats/
layout.rs
