sas7bdat 0.2.0

Rust library + CLI for decoding SAS7BDAT datasets and streaming them to modern 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

#[cfg(feature = "csv")]
mod csv;
#[cfg(feature = "parquet")]
mod parquet;

#[cfg(any(feature = "csv", feature = "parquet"))]
use crate::error::Error;
use crate::{
    cell::CellValue,
    dataset::DatasetMetadata,
    error::Result,
    parser::{ColumnInfo, ColumnarBatch, DatasetLayout, StreamingRow},
};
#[cfg(feature = "csv")]
pub use csv::CsvSink;
#[cfg(feature = "parquet")]
pub use parquet::ParquetSink;
#[cfg(any(feature = "csv", feature = "parquet"))]
use std::borrow::Cow;

/// Provides high-level dataset information to sinks during initialisation.
pub struct SinkContext<'a> {
    pub metadata: &'a DatasetMetadata,
    pub columns: &'a [ColumnInfo],
    pub source_path: Option<String>,
}

impl<'a> SinkContext<'a> {
    #[must_use]
    pub fn new(parsed: &'a DatasetLayout) -> Self {
        Self {
            metadata: &parsed.header.metadata,
            columns: &parsed.columns,
            source_path: None,
        }
    }
}

/// Trait implemented by row sinks that consume decoded SAS rows.
pub trait RowSink {
    /// Called before any rows are written to allow the sink to initialise internal state.
    ///
    /// # Errors
    ///
    /// Returns an error if the sink cannot be initialised for the provided metadata.
    fn begin(&mut self, context: SinkContext<'_>) -> Result<()>;

    /// Invoked for every decoded row produced by the parser.
    ///
    /// # Errors
    ///
    /// Returns an error if the row cannot be encoded or written to the underlying output.
    fn write_row(&mut self, row: &[CellValue<'_>]) -> Result<()>;

    /// Invoked for every decoded row when using the zero-copy streaming pipeline.
    ///
    /// The default implementation materialises the row into a temporary `Vec<CellValue>`
    /// before delegating to [`write_row`](RowSink::write_row).
    ///
    /// # Errors
    ///
    /// Propagates errors from materialisation or `write_row`.
    fn write_streaming_row(&mut self, row: StreamingRow<'_, '_>) -> Result<()> {
        let values = row.materialize()?;
        self.write_row(&values)
    }

    /// Called once all rows have been forwarded to the sink.
    ///
    /// # Errors
    ///
    /// Returns an error if finalising the sink or flushing the underlying output fails.
    fn finish(&mut self) -> Result<()>;
}

/// Trait implemented by sinks that can consume columnar batches directly.
pub trait ColumnarSink: RowSink {
    /// Writes a batch of rows that may be filtered via `selection`, which maps sink columns
    /// to their corresponding source column indices.
    ///
    /// # Errors
    ///
    /// Returns an error if decoding fails or if the column selection is invalid.
    fn write_columnar_batch(
        &mut self,
        batch: &ColumnarBatch<'_>,
        selection: &[usize],
    ) -> Result<()>;
}

#[cfg(any(feature = "csv", feature = "parquet"))]
pub(crate) fn validate_sink_begin(
    context: &SinkContext<'_>,
    writer_present: bool,
    sink_name: &str,
) -> Result<()> {
    if writer_present {
        return Err(Error::Unsupported {
            feature: Cow::Owned(format!(
                "{sink_name} sink cannot be reused without finishing"
            )),
        });
    }
    if context.metadata.variables.len() != context.columns.len() {
        return Err(Error::InvalidMetadata {
            details: Cow::from("column metadata length mismatch"),
        });
    }
    Ok(())
}