slow5lib/

writer.rs

use std::collections::BTreeSet;
use std::fs::File;
use std::io::{self, BufWriter, Write};
use std::path::Path;
use std::sync::{Arc, mpsc};

const BUF_SIZE: usize = 512 * 1024;

use crate::Result;
use crate::error::SlowError;
use crate::header::{Header, RecordCompression, SignalCompression};
use crate::record::Record;

/// Primary-field type header line written into every BLOW5 ASCII section.
const TYPE_HEADER: &str = "#char*\tuint32_t\tdouble\tdouble\tdouble\tdouble\tuint64_t\tint16_t*\n";

/// Primary-field column name header line written into every BLOW5 ASCII section.
const COL_HEADER: &str = "#read_id\tread_group\tdigitisation\toffset\trange\tsampling_rate\tlen_raw_signal\traw_signal\n";

/// Sequential writer for SLOW5 and BLOW5 files.
///
/// Records are written in the order they are passed to `write()`. The format
/// is detected from the path extension (`.slow5` or `.blow5`).
///
/// For high-throughput BLOW5 writing, compress records in parallel with rayon and
/// call `write()` in sorted order -- the writer itself must remain on one
/// thread (it owns the file cursor).
///
/// ```no_run
/// # use slow5lib::{Slow5Writer, header::{Header, RecordCompression, SignalCompression, ReadGroup}, aux::AuxMeta};
/// # fn example() -> slow5lib::Result<()> {
/// let header = Header {
///     version: (0, 2, 0),
///     num_read_groups: 1,
///     record_compression: RecordCompression::Zstd,
///     signal_compression: SignalCompression::SvbZd,
///     read_groups: vec![ReadGroup::default()],
///     aux_meta: AuxMeta::default(),
/// };
/// let mut writer = Slow5Writer::create("out.blow5", header)?;
/// // writer.write(&record)?;
/// writer.finish()?;
/// # Ok(()) }
/// ```
pub struct Slow5Writer {
    state: WriterState,
}

enum WriterState {
    Blow5 {
        inner: BufWriter<File>,
        record_compression: RecordCompression,
        signal_compression: SignalCompression,
        aux_meta: crate::aux::AuxMeta,
        // Reused across write() calls to avoid per-record ZSTD_CCtx alloc/free.
        // None when record_compression is not Zstd.
        zstd_compressor: Option<zstd::bulk::Compressor<'static>>,
    },
    Slow5 {
        inner: BufWriter<File>,
        aux_meta: crate::aux::AuxMeta,
    },
}

impl Slow5Writer {
    /// Create a new SLOW5 or BLOW5 file and write its header.
    ///
    /// The format is detected from the path extension (`.slow5` or `.blow5`).
    /// For BLOW5 with signal compression, set `header.version` to at least `(0, 2, 0)`.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use slow5lib::{Slow5Writer, header::{Header, RecordCompression, SignalCompression, ReadGroup}};
    /// use slow5lib::aux::AuxMeta;
    ///
    /// # fn example() -> slow5lib::Result<()> {
    /// let header = Header {
    ///     version: (0, 2, 0),
    ///     num_read_groups: 1,
    ///     record_compression: RecordCompression::Zstd,
    ///     signal_compression: SignalCompression::SvbZd,
    ///     read_groups: vec![ReadGroup::default()],
    ///     aux_meta: AuxMeta::default(),
    /// };
    /// let mut writer = Slow5Writer::create("out.blow5", header)?;
    /// // writer.write(&record)?;
    /// writer.finish()?;
    /// # Ok(()) }
    /// ```
    pub fn create(path: impl AsRef<Path>, header: Header) -> Result<Self> {
        let path = path.as_ref();

        let ext = path.extension().and_then(|e| e.to_str()).unwrap_or("");
        match ext {
            "blow5" => Self::create_blow5(path, header),
            "slow5" => Self::create_slow5(path, header),
            other => Err(SlowError::InvalidFormat(format!(
                "unrecognised extension '.{other}': expected '.blow5' or '.slow5'"
            ))),
        }
    }

    fn create_blow5(path: &Path, header: Header) -> Result<Self> {
        let file = File::create(path)?;
        let mut inner = BufWriter::with_capacity(BUF_SIZE, file);

        let header_bytes = build_blow5_header(&header)?;
        inner.write_all(&header_bytes)?;

        let zstd_compressor = if matches!(header.record_compression, RecordCompression::Zstd) {
            let c = zstd::bulk::Compressor::new(1)
                .map_err(|e| SlowError::Decompression(e.to_string()))?;
            Some(c)
        } else {
            None
        };

        Ok(Self {
            state: WriterState::Blow5 {
                inner,
                record_compression: header.record_compression,
                signal_compression: header.signal_compression,
                aux_meta: header.aux_meta,
                zstd_compressor,
            },
        })
    }

    fn create_slow5(path: &Path, header: Header) -> Result<Self> {
        let file = File::create(path)?;
        let mut inner = BufWriter::with_capacity(BUF_SIZE, file);

        crate::slow5::write_header(&mut inner, &header)?;

        Ok(Self {
            state: WriterState::Slow5 {
                inner,
                aux_meta: header.aux_meta,
            },
        })
    }

    /// Compress and write a single record.
    ///
    /// The signal is encoded and compressed according to the compression settings
    /// specified in the header at construction time.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use slow5lib::{Slow5Writer, record::Record};
    /// use slow5lib::header::{Header, RecordCompression, SignalCompression, ReadGroup};
    /// use slow5lib::aux::AuxMeta;
    /// use std::collections::HashMap;
    ///
    /// # fn example() -> slow5lib::Result<()> {
    /// let header = Header {
    ///     version: (0, 2, 0),
    ///     num_read_groups: 1,
    ///     record_compression: RecordCompression::Zstd,
    ///     signal_compression: SignalCompression::SvbZd,
    ///     read_groups: vec![ReadGroup::default()],
    ///     aux_meta: AuxMeta::default(),
    /// };
    /// let mut writer = Slow5Writer::create("out.blow5", header)?;
    ///
    /// let record = Record {
    ///     read_id: "my-read-001".to_string(),
    ///     read_group: 0,
    ///     digitisation: 2048.0,
    ///     offset: -224.0,
    ///     range: 1463.0,
    ///     sampling_rate: 4000.0,
    ///     raw_signal: vec![512i16, 515, 510, 518, 511],
    ///     aux: HashMap::new(),
    /// };
    /// writer.write(&record)?;
    /// writer.finish()?;
    /// # Ok(()) }
    /// ```
    pub fn write(&mut self, record: &Record) -> Result<()> {
        match &mut self.state {
            WriterState::Blow5 {
                inner,
                record_compression,
                signal_compression,
                aux_meta,
                zstd_compressor,
            } => {
                let record_buf = encode_record(record, *signal_compression, aux_meta)?;

                let (size_bytes, payload) = match record_compression {
                    RecordCompression::None => {
                        let size = record_buf.len() as u64;
                        (size.to_le_bytes(), record_buf)
                    }
                    RecordCompression::Zstd => {
                        let compressor = zstd_compressor.as_mut().expect("zstd compressor present");
                        let compressed = compressor
                            .compress(&record_buf)
                            .map_err(|e| SlowError::Decompression(e.to_string()))?;
                        let size = compressed.len() as u64;
                        (size.to_le_bytes(), compressed)
                    }
                    RecordCompression::Zlib => {
                        let compressed = crate::compression::compress_record_zlib(&record_buf)?;
                        let size = compressed.len() as u64;
                        (size.to_le_bytes(), compressed)
                    }
                };

                inner.write_all(&size_bytes)?;
                inner.write_all(&payload)?;
                Ok(())
            }
            WriterState::Slow5 { inner, aux_meta } => {
                crate::slow5::write_record(inner, record, aux_meta)
            }
        }
    }

    /// Compress and write a batch of records in parallel, then flush sequentially.
    ///
    /// All records are encoded and compressed concurrently across the rayon thread pool.
    /// The compressed payloads are then written to disk in order on the calling thread.
    /// Compression and I/O are not overlapped -- for that, see the manual `records_raw` pattern.
    ///
    /// For SLOW5 text files, this falls back to sequential `write()` calls since there
    /// is no signal compression to parallelize.
    ///
    /// Requires the `rayon` feature.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use slow5lib::{Slow5Writer, record::Record};
    /// use slow5lib::header::{Header, RecordCompression, SignalCompression, ReadGroup};
    /// use slow5lib::aux::AuxMeta;
    /// use std::collections::HashMap;
    ///
    /// # fn example() -> slow5lib::Result<()> {
    /// let header = Header {
    ///     version: (0, 2, 0),
    ///     num_read_groups: 1,
    ///     record_compression: RecordCompression::Zstd,
    ///     signal_compression: SignalCompression::SvbZd,
    ///     read_groups: vec![ReadGroup::default()],
    ///     aux_meta: AuxMeta::default(),
    /// };
    /// let mut writer = Slow5Writer::create("out.blow5", header)?;
    ///
    /// let records: Vec<Record> = Vec::new(); // normally populated from a reader
    /// writer.write_all_par(&records)?;
    /// writer.finish()?;
    /// # Ok(()) }
    /// ```
    #[cfg(feature = "rayon")]
    pub fn write_all_par(&mut self, records: &[Record]) -> Result<()> {
        use rayon::prelude::*;

        // SLOW5 text -- no record compression, fall back to sequential
        if matches!(self.state, WriterState::Slow5 { .. }) {
            for record in records {
                self.write(record)?;
            }
            return Ok(());
        }

        // Extract parameters; releases the immutable borrow before the mutable write below
        let (rc, sc, am) = match &self.state {
            WriterState::Blow5 {
                record_compression,
                signal_compression,
                aux_meta,
                ..
            } => (*record_compression, *signal_compression, aux_meta.clone()),
            WriterState::Slow5 { .. } => unreachable!(),
        };

        // Parallel: encode + compress
        let chunks: Vec<Result<(u64, Vec<u8>)>> = records
            .par_iter()
            .map(|record| {
                let record_buf = encode_record(record, sc, &am)?;
                compress_chunk(&record_buf, rc)
            })
            .collect();

        // Sequential: write to file in order
        let WriterState::Blow5 { inner, .. } = &mut self.state else {
            unreachable!()
        };
        for chunk in chunks {
            let (size, payload) = chunk?;
            inner.write_all(&size.to_le_bytes())?;
            inner.write_all(&payload)?;
        }
        Ok(())
    }

    /// Write the EOF marker (BLOW5) or flush (SLOW5) and close the file.
    ///
    /// Consumes `self` so the writer cannot be used after finishing.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use slow5lib::{Slow5Writer, header::{Header, RecordCompression, SignalCompression, ReadGroup}};
    /// use slow5lib::aux::AuxMeta;
    ///
    /// # fn example() -> slow5lib::Result<()> {
    /// let header = Header {
    ///     version: (0, 2, 0),
    ///     num_read_groups: 1,
    ///     record_compression: RecordCompression::None,
    ///     signal_compression: SignalCompression::None,
    ///     read_groups: vec![ReadGroup::default()],
    ///     aux_meta: AuxMeta::default(),
    /// };
    /// let writer = Slow5Writer::create("out.blow5", header)?;
    /// writer.finish()?; // flushes and writes EOF marker
    /// # Ok(()) }
    /// ```
    pub fn finish(self) -> Result<()> {
        match self.state {
            WriterState::Blow5 { mut inner, .. } => {
                // aux_meta consumed via `..`
                inner.write_all(crate::blow5::EOF_MARKER)?;
                inner.flush().map_err(SlowError::Io)
            }
            WriterState::Slow5 { mut inner, .. } => inner.flush().map_err(SlowError::Io),
        }
    }
}

// ── ParallelSlow5Writer ───────────────────────────────────────────────────────

/// Cloneable write handle for [`ParallelSlow5Writer`].
///
/// Obtain one from [`ParallelSlow5Writer::write_handle`] and clone it for each
/// rayon worker (or any thread). Calling `write()` compresses on the calling thread
/// and enqueues the payload for the background I/O thread. Blocks under backpressure.
///
/// Drop all handles before calling [`ParallelSlow5Writer::finish`].
///
/// # Examples
///
/// ```no_run
/// use slow5lib::{Slow5WriteHandle, writer::ParallelSlow5Writer};
/// use slow5lib::header::{Header, RecordCompression, SignalCompression, ReadGroup};
/// use slow5lib::aux::AuxMeta;
///
/// # fn example() -> slow5lib::Result<()> {
/// let header = Header {
///     version: (0, 2, 0),
///     num_read_groups: 1,
///     record_compression: RecordCompression::Zstd,
///     signal_compression: SignalCompression::SvbZd,
///     read_groups: vec![ReadGroup::default()],
///     aux_meta: AuxMeta::default(),
/// };
/// let par_writer = ParallelSlow5Writer::create("out.blow5", header, 32)?;
/// let handle = par_writer.write_handle();
/// // handle.write(&rec)?;
/// drop(handle);
/// par_writer.finish()?;
/// # Ok(()) }
/// ```
#[derive(Clone)]
pub struct Slow5WriteHandle {
    tx: mpsc::SyncSender<(u64, Vec<u8>)>,
    record_compression: RecordCompression,
    signal_compression: SignalCompression,
    aux_meta: Arc<crate::aux::AuxMeta>,
}

impl Slow5WriteHandle {
    /// Encode and compress `record` on the calling thread, then enqueue for writing.
    ///
    /// Blocks when the internal channel is full (backpressure from the I/O thread).
    /// Returns an error if the background writer thread has exited unexpectedly.
    pub fn write(&self, record: &crate::record::Record) -> Result<()> {
        let buf = encode_record(record, self.signal_compression, &self.aux_meta)?;
        let (size, payload) = compress_chunk(&buf, self.record_compression)?;
        self.tx.send((size, payload)).map_err(|_| {
            SlowError::Io(io::Error::new(
                io::ErrorKind::BrokenPipe,
                "writer thread has exited",
            ))
        })
    }
}

/// BLOW5 writer backed by a dedicated background I/O thread.
///
/// Rayon workers (or any threads) call `write()` on a cloned [`Slow5WriteHandle`] --
/// compression happens on the calling thread in parallel. The background thread does
/// all sequential file I/O, overlapping with compression work.
///
/// Records are written in the order they arrive at the background thread, which is
/// non-deterministic under parallel use. The BLOW5 format is order-independent; the
/// sidecar index provides random access regardless of record order.
///
/// # Usage pattern
///
/// ```no_run
/// use slow5lib::writer::ParallelSlow5Writer;
/// use slow5lib::header::{Header, RecordCompression, SignalCompression, ReadGroup};
/// use slow5lib::aux::AuxMeta;
///
/// # fn example() -> slow5lib::Result<()> {
/// let header = Header {
///     version: (0, 2, 0),
///     num_read_groups: 1,
///     record_compression: RecordCompression::Zstd,
///     signal_compression: SignalCompression::SvbZd,
///     read_groups: vec![ReadGroup::default()],
///     aux_meta: AuxMeta::default(),
/// };
/// let n_threads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4);
/// let par_writer = ParallelSlow5Writer::create("out.blow5", header, n_threads * 4)?;
/// let handle = par_writer.write_handle();
///
/// // rayon::prelude::*;
/// // ids.par_iter().for_each(|id| {
/// //     handle.write(&reader.get(id).unwrap()).unwrap();
/// // });
///
/// drop(handle);       // signal EOF -- all senders must be dropped before finish()
/// par_writer.finish() // join background thread, propagate any I/O error
/// # }
/// ```
pub struct ParallelSlow5Writer {
    tx: Option<mpsc::SyncSender<(u64, Vec<u8>)>>,
    handle: Option<std::thread::JoinHandle<Result<()>>>,
    record_compression: RecordCompression,
    signal_compression: SignalCompression,
    aux_meta: Arc<crate::aux::AuxMeta>,
}

impl ParallelSlow5Writer {
    /// Create a BLOW5 file and spawn the background writer thread.
    ///
    /// Only `.blow5` files are supported; returns an error for `.slow5` paths.
    ///
    /// `channel_cap` is the maximum number of compressed payloads buffered between
    /// compression threads and the writer thread. A value of `num_threads * 4` is a
    /// reasonable default. Larger values consume more memory; smaller values add
    /// backpressure earlier.
    pub fn create(path: impl AsRef<Path>, header: Header, channel_cap: usize) -> Result<Self> {
        let path = path.as_ref();
        let ext = path.extension().and_then(|e| e.to_str()).unwrap_or("");
        if ext != "blow5" {
            return Err(SlowError::InvalidFormat(
                "ParallelSlow5Writer only supports .blow5 files".into(),
            ));
        }

        let file = File::create(path)?;
        let mut inner = BufWriter::with_capacity(BUF_SIZE, file);
        let header_bytes = build_blow5_header(&header)?;
        inner.write_all(&header_bytes)?;

        let record_compression = header.record_compression;
        let signal_compression = header.signal_compression;
        let aux_meta = Arc::new(header.aux_meta);

        let (tx, rx) = mpsc::sync_channel::<(u64, Vec<u8>)>(channel_cap);

        let handle = std::thread::spawn(move || -> Result<()> {
            for (size, payload) in rx {
                inner
                    .write_all(&size.to_le_bytes())
                    .map_err(SlowError::Io)?;
                inner.write_all(&payload).map_err(SlowError::Io)?;
            }
            inner
                .write_all(crate::blow5::EOF_MARKER)
                .map_err(SlowError::Io)?;
            inner.flush().map_err(SlowError::Io)
        });

        Ok(Self {
            tx: Some(tx),
            handle: Some(handle),
            record_compression,
            signal_compression,
            aux_meta,
        })
    }

    /// Return a cloneable handle for submitting records from worker threads.
    ///
    /// Clone the handle once per rayon worker (or thread). All clones must be
    /// dropped before calling `finish()`.
    pub fn write_handle(&self) -> Slow5WriteHandle {
        Slow5WriteHandle {
            tx: self.tx.as_ref().expect("writer already finished").clone(),
            record_compression: self.record_compression,
            signal_compression: self.signal_compression,
            aux_meta: Arc::clone(&self.aux_meta),
        }
    }

    /// Close the channel and join the background thread, returning any I/O error.
    ///
    /// **All `Slow5WriteHandle` clones must be dropped before calling this.**
    /// If any handle is still live, the background thread will keep waiting for
    /// records and this call will block indefinitely.
    pub fn finish(mut self) -> Result<()> {
        drop(self.tx.take());
        let h = self.handle.take().expect("writer already finished");
        match h.join() {
            Ok(result) => result,
            Err(_) => Err(SlowError::Io(io::Error::other("writer thread panicked"))),
        }
    }
}

impl Drop for ParallelSlow5Writer {
    fn drop(&mut self) {
        drop(self.tx.take());
        if let Some(h) = self.handle.take() {
            let _ = h.join();
        }
    }
}

/// Compress one encoded record buffer and return `(compressed_size, payload)`.
fn compress_chunk(buf: &[u8], rc: RecordCompression) -> Result<(u64, Vec<u8>)> {
    match rc {
        RecordCompression::None => Ok((buf.len() as u64, buf.to_vec())),
        RecordCompression::Zstd => {
            let c = crate::compression::compress_record_zstd_tl(buf)?;
            Ok((c.len() as u64, c))
        }
        RecordCompression::Zlib => {
            let c = crate::compression::compress_record_zlib(buf)?;
            Ok((c.len() as u64, c))
        }
    }
}

/// Build the complete BLOW5 binary header as a byte buffer.
///
/// Binary wire layout:
/// ```text
/// [magic: 6][major: 1][minor: 1][patch: 1][rec_comp: 1][num_read_groups: 4]
/// [sig_comp: 1 if version >= 0.2.0][padding: zeros to offset 64]
/// [header_size: u32 le][ASCII section: header_size bytes]
/// ```
///
/// `header_size` is the byte count of the ASCII section (not including its own
/// 4-byte field). It is computed after building the ASCII section and patched
/// back into the buffer at offset 64.
fn build_blow5_header(header: &Header) -> Result<Vec<u8>> {
    let mut buf: Vec<u8> = Vec::with_capacity(256);

    // Fixed binary prefix
    buf.extend_from_slice(crate::blow5::MAGIC);
    buf.push(header.version.0);
    buf.push(header.version.1);
    buf.push(header.version.2);
    buf.push(header.record_compression.to_byte());
    buf.extend_from_slice(&header.num_read_groups.to_le_bytes());

    // Signal compression byte present only for version >= 0.2.0
    if (header.version.0, header.version.1) >= (0, 2) {
        buf.push(header.signal_compression.to_byte());
    }

    // Pad with zeros to reach offset 64
    while buf.len() < 64 {
        buf.push(0);
    }

    // Placeholder for header_size (u32 le) at byte 64
    let header_size_pos = buf.len(); // == 64
    buf.extend_from_slice(&0u32.to_le_bytes());

    // ASCII section begins at byte 68
    let ascii_start = buf.len(); // == 68

    // @attribute lines -- collect all unique attribute names sorted for determinism
    let all_keys: BTreeSet<&str> = header
        .read_groups
        .iter()
        .flat_map(|rg| rg.attributes.keys().map(String::as_str))
        .collect();

    for key in &all_keys {
        buf.push(b'@');
        buf.extend_from_slice(key.as_bytes());
        for rg in &header.read_groups {
            buf.push(b'\t');
            let val = rg.attributes.get(*key).map(String::as_str).unwrap_or("");
            if val.is_empty() {
                buf.push(b'.');
            } else {
                buf.extend_from_slice(val.as_bytes());
            }
        }
        buf.push(b'\n');
    }

    // Type header line: primary fields + aux fields
    buf.extend_from_slice(TYPE_HEADER.trim_end_matches('\n').as_bytes());
    for typ in &header.aux_meta.types {
        buf.push(b'\t');
        buf.extend_from_slice(crate::aux::aux_type_str(typ).as_bytes());
    }
    buf.push(b'\n');

    // Column name header line: primary fields + aux fields
    buf.extend_from_slice(COL_HEADER.trim_end_matches('\n').as_bytes());
    for name in &header.aux_meta.names {
        buf.push(b'\t');
        buf.extend_from_slice(name.as_bytes());
    }
    buf.push(b'\n');

    // Patch header_size with the actual ASCII byte count
    let ascii_len = (buf.len() - ascii_start) as u32;
    buf[header_size_pos..header_size_pos + 4].copy_from_slice(&ascii_len.to_le_bytes());

    Ok(buf)
}

/// Encode a `Record` into the uncompressed binary record buffer.
///
/// Wire layout (before record compression):
/// ```text
/// [read_id_len: u16 le][read_id: N bytes]
/// [read_group: u32 le]
/// [digitisation: f64 le][offset: f64 le][range: f64 le][sampling_rate: f64 le]
/// [len_raw_signal: u64 le][signal bytes]
/// ```
///
/// For `SignalCompression::None`: `len_raw_signal` = sample count; signal bytes = i16 LE samples.
/// For `SignalCompression::SvbZd`: `len_raw_signal` = byte count of SVB-ZD data; signal bytes = compressed.
/// For `SignalCompression::ExZd`: `len_raw_signal` = byte count of the ex-zd frame; signal bytes = compressed.
fn encode_record(
    record: &Record,
    signal_compression: SignalCompression,
    aux_meta: &crate::aux::AuxMeta,
) -> Result<Vec<u8>> {
    let read_id_bytes = record.read_id.as_bytes();
    let read_id_len = u16::try_from(read_id_bytes.len()).map_err(|_| {
        SlowError::InvalidFormat(format!(
            "read_id '{}' is too long ({} bytes, max 65535)",
            record.read_id,
            read_id_bytes.len()
        ))
    })?;

    let mut buf =
        Vec::with_capacity(2 + read_id_bytes.len() + 4 + 8 * 5 + 2 * record.raw_signal.len());

    buf.extend_from_slice(&read_id_len.to_le_bytes());
    buf.extend_from_slice(read_id_bytes);
    buf.extend_from_slice(&record.read_group.to_le_bytes());
    buf.extend_from_slice(&record.digitisation.to_le_bytes());
    buf.extend_from_slice(&record.offset.to_le_bytes());
    buf.extend_from_slice(&record.range.to_le_bytes());
    buf.extend_from_slice(&record.sampling_rate.to_le_bytes());

    match signal_compression {
        SignalCompression::None => {
            let n_samples = record.raw_signal.len() as u64;
            buf.extend_from_slice(&n_samples.to_le_bytes());
            for &s in &record.raw_signal {
                buf.extend_from_slice(&s.to_le_bytes());
            }
        }
        SignalCompression::SvbZd => {
            let compressed = crate::compression::compress_signal_svb_zd(&record.raw_signal)?;
            let byte_count = compressed.len() as u64;
            buf.extend_from_slice(&byte_count.to_le_bytes());
            buf.extend_from_slice(&compressed);
        }
        SignalCompression::ExZd => {
            let compressed = crate::compression::compress_signal_ex_zd(&record.raw_signal)?;
            let byte_count = compressed.len() as u64;
            buf.extend_from_slice(&byte_count.to_le_bytes());
            buf.extend_from_slice(&compressed);
        }
    }

    // Auxiliary fields in schema order
    for (name, typ) in aux_meta.names.iter().zip(aux_meta.types.iter()) {
        let val = record
            .aux
            .get(name)
            .unwrap_or(&crate::aux::AuxValue::Missing);
        crate::aux::encode_binary(val, typ, &mut buf);
    }

    Ok(buf)
}