Struct ParquetLogger 

pub(crate) struct ParquetLogger {
    pub(crate) writer: SerializedFileWriter<File>,
    pub(crate) inprogress_path: PathBuf,
    pub(crate) final_path: PathBuf,
    pub(crate) timestamps: Vec<i64>,
    pub(crate) values: Vec<f64>,
    pub(crate) max_rows: usize,
    pub(crate) first_timestamp_unix_ns: Option<i64>,
    pub(crate) row_groups_written: usize,
    pub(crate) date: NaiveDate,
    pub(crate) asset: u32,
    pub(crate) channel: u8,
    pub(crate) file_index: usize,
}

Buffered writer for one channel’s Parquet file.

Rows are buffered to form row groups. Files are partitioned by asset, UTC date, channel, and monotonically increasing part index. While open, the file is named part-<NNNN>.parquet.inprogress; Self::close renames it to part-<NNNN>.parquet.

Fields

writer: SerializedFileWriter<File>

Parquet writer over the .inprogress file.

inprogress_path: PathBuf

Path of the file while it is being written.

final_path: PathBuf

Path the file is renamed to when it is closed.

timestamps: Vec<i64>

Buffered timestamps (Unix nanoseconds) of the current row group.

values: Vec<f64>

Buffered sample values of the current row group, parallel to timestamps.

max_rows: usize

Row count at which the buffer is written as a row group.

first_timestamp_unix_ns: Option<i64>

Timestamp (Unix nanoseconds) of the first row written to this file.

row_groups_written: usize

Number of row groups written to the file so far.

date: NaiveDate

UTC date of the samples in this file.

asset: u32

Asset number the file belongs to.

channel: u8

LabJack channel the file belongs to.

file_index: usize

Part index in the file name.

Implementations

impl ParquetLogger

pub(crate) fn new( asset: u32, channel: u8, file_index: usize, date: NaiveDate, calibration: CalibrationSpec, parquet_root: &Path, ) -> Self

Creates a new Parquet file for one asset, channel, date, and part index.

The file is created at <parquet_root>/asset<NNN>/<YYYY-MM-DD>/ch<NN>/part-<NNNN>.parquet.inprogress, creating directories as needed. The calibration spec is written into file-level key-value metadata (key calibration) so exported data can be interpreted without consulting the live config. If the spec cannot be serialized, {} is stored instead.

Arguments

• asset - Asset number for the asset<NNN> directory.
• channel - LabJack channel for the ch<NN> directory.
• file_index - Part index for the file name, usually from next_file_index.
• date - UTC date of the samples, for the date directory.
• calibration - Calibration stored in the file metadata.
• parquet_root - Root output directory.

Panics

Panics if the directory or file cannot be created, or if the Parquet writer cannot be set up.

pub(crate) fn write_run( &mut self, first_timestamp_unix_ns: i64, sample_interval_ns: u64, rotate_secs: u64, values: impl ExactSizeIterator<Item = f64>, ) -> Result<usize, Box<dyn Error + Send + Sync>>

Buffers the leading samples of a run that belong in this file.

The run’s samples are at first_timestamp_unix_ns, first_timestamp_unix_ns + sample_interval_ns and so on, one per value. The first one must belong in this file (see Self::accepts). The samples are buffered up to the first one that belongs in another file or the row group limit, whichever comes first (see Self::run_len), and the row group is flushed if it is full. This writes the same rows and row groups as buffering the samples one by one with a rotation check before each. The first call also records the file’s first timestamp, which Self::should_rotate_before uses to find the file’s rotation window.

Arguments

• first_timestamp_unix_ns - Time of the first value, in Unix nanoseconds.
• sample_interval_ns - Time between values.
• rotate_secs - Rotation window length in seconds.
• values - Raw sample values, in time order. Every one of their timestamps must fit in an i64.

Returns

How many values were buffered: at least 1 unless values is empty.

Errors

Returns an error if writing the full row group fails (see Self::flush). The file is then unusable and should be dropped without closing it.

pub(crate) fn accepts(&self, timestamp_unix_ns: i64, rotate_secs: u64) -> bool

Returns whether a sample belongs in this file.

It does when it has the file’s UTC date and Self::should_rotate_before is false.

pub(crate) fn run_len( &self, timestamp_unix_ns: i64, sample_interval_ns: u64, max_len: usize, rotate_secs: u64, ) -> usize

Counts how many samples of a run can be written to this file in one go.

The run starts at timestamp_unix_ns, which this file must accept (see Self::accepts), and rises by sample_interval_ns per sample. The file’s first timestamp must already be recorded. Timestamps only rise, and the file accepts every timestamp from its first one up to the end of its rotation window or UTC date, so the accepted samples are a prefix of the run. The count is also capped at the rows left before the row group is full, so a flush (which can make Self::should_rotate_before change its answer) always ends a run.

Checks the last sample first, which settles the common case of a whole message inside one window, and otherwise binary searches for the first rejected sample.

Arguments

• timestamp_unix_ns - Time of the run’s first sample.
• sample_interval_ns - Time between samples.
• max_len - Samples left in the run. Every one of their timestamps must fit in an i64.
• rotate_secs - Rotation window length in seconds.

Returns

A count between 1 and max_len, or 0 when max_len is 0.

pub(crate) fn flush(&mut self) -> Result<(), Box<dyn Error + Send + Sync>>

Writes the current buffer as a Parquet row group.

Does nothing when the buffer is empty. Otherwise writes the timestamp and value columns, closes the row group and clears the buffer. The data is not synced to disk here; that happens in Self::close.

Errors

Returns an error if the Parquet writer fails to start, write or close the row group or one of its columns. The buffer is left as it was, and the file should be treated as broken.

pub(crate) fn should_rotate_before( &self, timestamp_unix_ns: i64, rotate_secs: u64, ) -> bool

Returns whether the next source sample belongs in a new file.

Files cover aligned windows of source time (see rotation_window). Using source time rather than wall-clock time keeps backlog replay correct: several hours of samples can be consumed in a few seconds.

Arguments

• timestamp_unix_ns - Time of the next sample, in Unix nanoseconds.
• rotate_secs - Rotation window length in seconds.

Returns

true if the file already has MAX_ROW_GROUPS_PER_FILE row groups, if the sample is earlier than the file’s first sample, or if it falls in a different rotation window. false for an empty file otherwise.

pub(crate) fn close(self) -> Result<PathBuf, Box<dyn Error + Send + Sync>>

Flushes buffered rows, writes the footer, syncs, and publishes the file.

The data and the rename are both flushed to disk before returning, so a caller that acks JetStream messages afterwards cannot lose them to a power cut. The steps are: write remaining rows, write the footer, fsync the file, rename .parquet.inprogress to .parquet, then fsync the directory.

A failed directory fsync is logged but does not fail the close; see sync_directory_after_publish for why.

Returns

The final .parquet path.

Errors

Returns an error if writing the remaining rows or the footer, syncing the file, or renaming it fails. The file then keeps its .inprogress name, so it is never read and is quarantined at the next start.