The RecordParser Trait¶

The RecordParser trait turns a byte chunk into field/value events fed to a ColumnarSink. This is where format-specific parsing lives.

See Architecture for how the engine calls the parser and how the sink accumulates values.

Trait definition¶

pub trait RecordParser: Send + Sync {
    /// Validate that the whole byte slice is well-formed.
    fn validate(&self, bytes: &[u8]) -> Result<()>;

    /// Parse one chunk and feed all row events into `sink`.
    fn parse_chunk(&self, bytes: &[u8], sink: &mut dyn ColumnarSink) -> Result<()>;

    /// Parse one chunk with monomorphized sink (inlinable, devirtualized).
    /// Default delegates to parse_chunk.
    fn parse_chunk_generic<S: ColumnarSink>(&self, bytes: &[u8], sink: &mut S) -> Result<()>
    where Self: Sized;
}

Only validate and parse_chunk are required; parse_chunk_generic has a default implementation that delegates to parse_chunk. The full method signatures, including the where Self: Sized bound that makes the generic method object-safe to skip, are in the Rust API reference. For how the engine drives these methods across execution modes, see Architecture: Decoder.

validate¶

Called once per chunk before parsing. Use it for upfront checks like UTF-8 validation. This is cheap (SIMD-accelerated) and catches malformed input early.

fn validate(&self, bytes: &[u8]) -> Result<()> {
    simdutf8::basic::from_utf8(bytes).map_err(rypipe_core::Error::Utf8)?;
    Ok(())
}

parse_chunk¶

The main parsing loop. For each record in the chunk:

1. Call sink.begin_row() to start a new row
2. For each field, call sink.put_field(name, value) or faster alternatives
3. Call sink.end_row() to finalize the row

The engine calls this once per chunk. Your parser sees a contiguous byte range that starts and ends at record boundaries (guaranteed by the Splitter).

The put_field(name, value) form shown below is the simplest sink call, but it pays a hash lookup per field. The faster alternatives (put_field_resolved, put_field_at, raw variants) are covered in The ColumnarSink Trait and listed in the Rust API reference.

fn parse_chunk(&self, bytes: &[u8], sink: &mut dyn ColumnarSink) -> Result<()> {
    let text = std::str::from_utf8(bytes)
        .map_err(|e| rypipe_core::Error::Plan(e.to_string()))?;
    for line in text.lines() {
        if line.is_empty() { continue; }
        sink.begin_row();
        for (col, value) in self.header.iter().zip(line.split(',')) {
            sink.put_field(col, Value::Str(Cow::Borrowed(value)));
        }
        sink.end_row();
    }
    Ok(())
}

parse_chunk_generic¶

The generic version allows the compiler to devirtualize sink calls (no vtable dispatch). When the engine knows the concrete sink type (e.g., TableBuilder), it calls parse_chunk_generic instead of parse_chunk.

fn parse_chunk_generic<S: ColumnarSink>(&self, bytes: &[u8], sink: &mut S) -> Result<()> {
    // Same body as parse_chunk, but sink calls are devirtualized.
    self.parse_chunk(bytes, sink as &mut dyn ColumnarSink)
}

Override this method for a measurable speedup on hot paths. The compiler can then inline sink.begin_row(), sink.put_field(), and sink.end_row() into the parsing loop, eliminating vtable dispatch overhead. For why the engine calls the generic form on every execution path (columnar, parallel, and streaming), see Adapter design and Architecture: Decoder.

Performance tips¶

1. Use Cow::Borrowed for non-entity text¶

Borrow from the input buffer when possible:

sink.put_field(col, Value::Str(Cow::Borrowed(value)));

The buffered filter path may hold values past the end of your parse function, so a borrow of a temporary would dangle. Cow::Borrowed is safe because it borrows from the chunk's byte slice, which lives long enough.

2. Check wants() before expensive extraction¶

Skip dropped fields entirely:

if sink.wants(col) {
    // expensive extraction here
    sink.put_field(col, value);
}

When wants() returns false, the engine will drop the column. Checking before extraction saves the cost of parsing, decoding, and allocating.

3. Use resolve + put_field_resolved for expensive extraction¶

When extraction is costly (entity unescaping, base64 decode, date parsing), resolve once and push with the resolved name to avoid a second hash probe:

if let Some(resolved) = sink.resolve(col) {
    let decoded = expensive_decode(value);
    sink.put_field_resolved(resolved, Value::Str(Cow::Owned(decoded)));
}

This pays the rename→drop hash lookup once instead of twice (once in resolve_and_put's internal resolve, once in put_field's internal ensure_column_idx).

4. Emit typed Value variants when possible¶

// Instead of:
sink.put_field("amount", Value::Str(Cow::Borrowed("123.45")));

// Do:
sink.put_field("amount", Value::Float64(123.45));

Typed variants skip the string-to-number conversion in the engine. The engine still stores the value correctly and exports it as the right Arrow type. The full variant list, including how decimals work without a Decimal128 variant, is in Rust Adapter Creation: Value types.

5. Do not call end_row() for partial trailing rows¶

If your parser reaches the end of the chunk mid-record, just return. The engine discards partial trailing rows automatically during normalize().

6. Skip blank lines or use them as record boundaries¶

Blank lines in the input produce empty rows with all null fields if you don't handle them. There are two patterns:

Pattern A: Skip blank lines (one-line-per-record formats).

for line in text.lines() {
    if line.is_empty() { continue; }
    sink.begin_row();
    // ... parse fields ...
    sink.end_row();
}

Pattern B: Blank lines as record separators (multi-line formats).

If your format puts one field per line with blank-line-separated records, split on \n\n in the Splitter and accumulate fields across lines in the Parser. Call end_row() only when you hit a blank line:

let mut in_record = false;
for line in text.lines() {
    if line.is_empty() {
        if in_record { sink.end_row(); in_record = false; }
        continue;
    }
    if !in_record { sink.begin_row(); in_record = true; }
    // ... put_field for each line ...
}
if in_record { sink.end_row(); }

See Multi-line Record Adapter for a complete worked example.

7. Within-chunk state¶

The parser is stateless across chunks (each chunk is parsed on an independent thread). But it CAN hold state within a chunk (tracking [section] headers, accumulating fields across lines, or handling \ continuations:

struct PropertiesParser;

impl RecordParser for PropertiesParser {
    // ... validate, etc.

    fn parse_chunk(&self, bytes: &[u8], sink: &mut dyn ColumnarSink) -> Result<()> {
        let text = std::str::from_utf8(bytes)
            .map_err(|e| rypipe_core::Error::Plan(e.to_string()))?;

        let mut pending_key: Option<String> = None;
        let mut pending_value = String::new();

        for line in text.lines() {
            if line.is_empty() {
                // Flush any pending continuation
                if let Some(key) = pending_key.take() {
                    sink.begin_row();
                    if sink.wants("key") {
                        sink.put_field("key", Value::Str(Cow::Borrowed(&key)));
                    }
                    if sink.wants("value") {
                        sink.put_field("value", Value::Str(Cow::Owned(pending_value.clone())));
                    }
                    sink.end_row();
                    pending_value.clear();
                }
                continue;
            }

            // ... parse key=value, handle continuation, etc.
        }
        Ok(())
    }
}

The state (pending_key, pending_value) persists across lines within the chunk but is NOT shared between chunks. The Splitter ensures continuations don't span chunk boundaries (see Splitter: Stateful formats).

8. Consider parse_chunk_generic for hot paths¶

If your parser is called millions of times (e.g., small files, streaming), override parse_chunk_generic to get devirtualized sink calls:

fn parse_chunk_generic<S: ColumnarSink>(&self, bytes: &[u8], sink: &mut S) -> Result<()> {
    let text = std::str::from_utf8(bytes)
        .map_err(|e| rypipe_core::Error::Plan(e.to_string()))?;
    for line in text.lines() {
        if line.is_empty() { continue; }
        sink.begin_row();
        for (col, value) in self.header.iter().zip(line.split(',')) {
            if sink.wants(col) {
                sink.put_field(col, Value::Str(Cow::Borrowed(value)));
            }
        }
        sink.end_row();
    }
    Ok(())
}

The parsing lifecycle¶

validate(bytes)                    ← called once per chunk
  │
  ▼
parse_chunk(bytes, sink)           ← called once per chunk
  │
  ├─ sink.begin_row()              ← clear per-row state
  ├─ sink.put_field("a", val)      ← push field (engine resolves + stores)
  ├─ sink.put_field("b", val)      ← push field
  ├─ sink.end_row()                ← null-fill missing, evaluate filter
  ├─ sink.begin_row()              ← next row
  ├─ ...
  └─ sink.end_row()               ← last row
  │
  ▼
sink.finish()                      ← called once after all chunks
  │
  ▼
Arrow RecordBatch                  ← zero-copy export

The same flow with the engine's internals filled in (chunk planning, parallel export, merge paths) is diagrammed in Architecture: Data flow.

Error handling¶

Return Err from parse_chunk to abort parsing. The engine will propagate the error to the caller. Common error types:

• rypipe_core::Error::Utf8: invalid UTF-8 in input
• rypipe_core::Error::Plan: invalid plan or configuration
• rypipe_core::Error::Io: I/O error

The complete Error enum (including Merge, Parser, and Lifetime) is documented in the Rust API reference.

Do not panic in parse_chunk. Panics are caught by catch_unwind in the parallel executor, but they abort the entire parse.

Build and test¶

Run the parser unit tests: one feeds a two-row sample through TableBuilder and checks row and column counts, the other confirms validate() rejects invalid UTF-8 instead of panicking:

$ cargo test parser
running 2 tests
test tests::parser_rejects_invalid_utf8 ... ok
test tests::parser_emits_all_rows ... ok

test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 8 filtered out; finished in 0.00s

Both finish in microseconds, so use them as a smoke test after every change to parse_chunk. For the end-to-end check through Python, see the walkthrough.

What the end user sees¶

The wants() checks and typed Value variants in your parse_chunk pay off as free projection pushdown and typed columns. The user chains a filter stage and passes type hints; the parser skips dropped fields and the engine builds typed Arrow arrays directly:

from rypipe_log import LogSource, FilterRows

# The filter is evaluated row-by-row during parsing; "amount" arrives as float64.
table = (
    LogSource("sample.log", field_types={"amount": "float64"})
    | FilterRows(field="status", op="eq", value="active")
).to_arrow()