Rust Adapter Creation¶

pub trait Splitter: Send + Sync {
    /// Find the start of the next record after byte offset `from`.
    fn next_record_start(&self, bytes: &[u8], from: usize) -> Option<usize>;

    /// Estimate bytes per row from a sample of the file.
    fn estimate_bytes_per_row(&self, sample: &[u8]) -> usize;
}

fn next_record_start(&self, bytes: &[u8], from: usize) -> Option<usize> {
    memchr::memchr(b'\n', &bytes[from..])          // SIMD-accelerated scan
        .map(|r| from + r + 1)                      // offset past the newline
}

fn estimate_bytes_per_row(&self, sample: &[u8]) -> usize {
    let n = sample.iter().filter(|&&b| b == b'\n').count().max(1);
    (sample.len() / n).max(1)   // guard against division by zero
}

pub trait RecordParser: Send + Sync {
    /// Validate that the bytes are valid for this format.
    fn validate(&self, bytes: &[u8]) -> Result<()>;

    /// Parse a chunk of bytes into field/value events.
    fn parse_chunk(&self, bytes: &[u8], sink: &mut dyn ColumnarSink) -> Result<()>;
}

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

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 part in line.split(',') {
            if let Some((key, value)) = part.split_once('=') {
                if sink.wants(key) {                // skip dropped fields
                    sink.put_field(key, Value::Str(Cow::Borrowed(value)));
                }
            }
        }
        sink.end_row();
    }
    Ok(())
}

// Simpler: two hash probes per field
if sink.wants(name) { sink.put_field(name, value); }

// Faster: single hash probe: resolve returns the resolved column name
if let Some(resolved) = sink.resolve(name) {
    sink.put_field_resolved(resolved, value);
}

use std::borrow::Cow;
use rypipe_core::{Splitter, RecordParser, ColumnarSink, Value, Result};

// Splitter: tells the engine where records begin so it can carve the
// input into chunks for parallel parsing. Clone is required by Pipeline.
#[derive(Clone, Default)]
pub struct LogSplitter;

impl Splitter for LogSplitter {
    // Find the next record boundary after `from`. The log format is
    // newline-delimited, so a SIMD-accelerated newline scan suffices;
    // return the offset just past the newline.
    fn next_record_start(&self, bytes: &[u8], from: usize) -> Option<usize> {
        memchr::memchr(b'\n', &bytes[from..]).map(|r| from + r + 1)
    }

    // Rough average row size from a file sample. The engine uses this to
    // size chunks and memory budgets, so it only needs to be approximate.
    fn estimate_bytes_per_row(&self, sample: &[u8]) -> usize {
        let n = sample.iter().filter(|&&b| b == b'\n').count().max(1);
        (sample.len() / n).max(1)  // never return 0
    }
}

// Parser: turns each byte chunk into row/field events on the sink.
// Stateless here; formats with headers or options would store them in
// this struct.
#[derive(Clone, Default)]
pub struct LogParser;

impl RecordParser for LogParser {
    // Called once per chunk before parsing. Reject invalid input early
    // with a fast SIMD UTF-8 check.
    fn validate(&self, bytes: &[u8]) -> Result<()> {
        simdutf8::basic::from_utf8(bytes)
            .map_err(rypipe_core::Error::Utf8)?;
        Ok(())
    }

    // The hot path: iterate rows and emit begin_row / put_field / end_row
    // for each record.
    fn parse_chunk(&self, bytes: &[u8], sink: &mut dyn ColumnarSink) -> Result<()> {
        // validate() already ran, so this cannot fail on real input;
        // map the error anyway instead of unwrapping.
        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; }  // tolerate trailing newlines
            sink.begin_row();
            for part in line.split(',') {
                // Skip tokens without '=' (malformed fields) rather than
                // failing the whole chunk.
                if let Some((key, value)) = part.split_once('=') {
                    // wants() is false for fields the plan drops or renames
                    // away; skipping them saves real CPU.
                    if sink.wants(key) {
                        // Borrow from the input: zero-copy until the sink
                        // decides to materialize the value.
                        sink.put_field(key, Value::Str(Cow::Borrowed(value)));
                    }
                }
            }
            sink.end_row();
        }
        Ok(())
    }
}

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 cleaned = text.replace("tns:", "").replace("xs:", "");  // strip ns prefixes

    for line in cleaned.lines() {
        if line.contains(&self.row_tag) {
            sink.begin_row();
            self.parse_xml_row(line, sink)?;
            sink.end_row();
        }
    }
    Ok(())
}

fn parse_chunk(&self, bytes: &[u8], sink: &mut dyn ColumnarSink) -> Result<()> {
    let regions = self.find_skip_regions(bytes);  // pre-compute byte ranges to skip

    for line in bytes.split(|&b| b == b'\n') {
        if self.in_skip_region(line.as_ptr() as usize, &regions) { continue; }
        sink.begin_row();
        self.parse_line(line, sink)?;
        sink.end_row();
    }
    Ok(())
}

fn parse_field(&self, name: &str, value: &str, sink: &mut dyn ColumnarSink) {
    match name {
        "age" => {
            if let Ok(n) = value.parse::<i64>() { sink.put_field(name, Value::Int64(n)); }
        }
        "score" => {
            if let Ok(f) = value.parse::<f64>() { sink.put_field(name, Value::Float64(f)); }
        }
        "active" => {
            sink.put_field(name, Value::Bool(matches!(value, "true" | "1" | "yes")));
        }
        _ => { sink.put_field(name, Value::Str(Cow::Borrowed(value))); }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use rypipe_core::{ExecutionPlan, TableBuilder};
    use std::sync::Arc;

    const SAMPLE: &[u8] = b"name=Alice,age=30,active=true\nname=Bob,age=25,active=false\n";

    #[test]
    fn splitter_finds_row_starts() {
        let s = LogSplitter;
        assert_eq!(s.next_record_start(SAMPLE, 0), Some(30));
        assert_eq!(s.next_record_start(SAMPLE, 30), Some(SAMPLE.len()));
        assert_eq!(s.next_record_start(SAMPLE, SAMPLE.len()), None);
    }

    #[test]
    fn parser_emits_all_rows() {
        let plan: Arc<ExecutionPlan> = Arc::new(ExecutionPlan::new());
        let mut builder = TableBuilder::with_plan(1024, plan);
        LogParser.validate(SAMPLE).unwrap();
        LogParser.parse_chunk(SAMPLE, &mut builder).unwrap();
        let batch = builder.finish().unwrap();
        assert_eq!(batch.num_rows(), 2);
        assert_eq!(batch.num_columns(), 3);
    }
}

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(format!("invalid UTF-8: {e}")))?;

    for (i, line) in text.lines().enumerate() {
        if line.is_empty() { continue; }
        sink.begin_row();
        for part in line.split(',') {
            let (key, val) = part.split_once('=').ok_or_else(|| {
                rypipe_core::Error::Plan(format!("line {i}: missing '=' in field: {part}"))
            })?;
            if sink.wants(key) {
                sink.put_field(key, Value::Str(Cow::Borrowed(val)));
            }
        }
        sink.end_row();
    }
    Ok(())
}

let pos = memchr::memchr(b'<', bytes);   // Good: SIMD-accelerated
let pos = bytes.iter().position(|&b| b == b'<');  // Bad: 5-10x slower

use arrow::pyarrow::ToPyArrow;
use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;
use pyo3::types::PyModule;
use rypipe_core::{ExecutionPlan, FieldType, Pipeline};
use std::collections::HashMap;

// One function is the whole Python API: Python passes the merged pushdown
// plan as plain kwargs, everything after the boundary stays in Rust.
#[pyfunction]
// Every plan kwarg is optional; defaults match the Python side's defaults
// so `read_log(path)` alone does something sensible.
#[pyo3(signature = (path, field_mapping=None, drop_fields=None, field_types=None,
                    auto_dict=false, use_mmap=false, prefault=false))]
fn read_log(
    path: String,
    field_mapping: Option<HashMap<String, String>>,  // RenameFields pushdown
    drop_fields: Option<Vec<String>>,                // DropFields pushdown
    field_types: Option<HashMap<String, String>>,    // CastTypes pushdown
    auto_dict: bool,
    use_mmap: bool,
    prefault: bool,
) -> PyResult<Py<PyAny>> {
    // Translate the Python kwargs into the engine's ExecutionPlan.
    let mut plan = ExecutionPlan::new();
    if let Some(map) = field_mapping {
        plan.field_map = map.into_iter().collect();
    }
    if let Some(drop) = drop_fields {
        plan.drop_fields = drop.into_iter().collect();
    }
    plan.auto_dict = auto_dict;
    if let Some(ft) = field_types {
        for (name, type_str) in ft {
            // Type names arrive as strings ("int64", "float64", ...);
            // reject unknown ones with a Python ValueError naming both
            // the type and the field.
            let ft = type_str.parse::<FieldType>().map_err(|_| {
                PyValueError::new_err(format!("unknown field type '{type_str}' for '{name}'"))
            })?;
            plan.field_types.insert(name, ft);
        }
    }

    // Run the whole parse in Rust: split, parse, fuse the plan, merge
    // chunks; one RecordBatch comes out. Engine errors become ValueError.
    let batch = Pipeline::new(LogSplitter, LogParser)
        .with_plan(plan)
        .read_path(&path, use_mmap, prefault)
        .map_err(|e| PyValueError::new_err(e.to_string()))?;

    // Export to Python: convert the batch via Arrow's PyO3 bridge, then
    // wrap it in a pyarrow.Table (the type the Python side promises).
    Python::attach(|py| {
        let pa = PyModule::import(py, "pyarrow")?;
        let rb = batch.to_pyarrow(py)?;
        let table = pa
            .getattr("Table")?
            .call_method1("from_batches", (vec![rb],))?;
        Ok(table.into())
    })
}

// Module init: the name must match [tool.maturin] module-name
// (`rypipe_log._rypipe_log`).
#[pymodule]
fn _rypipe_log(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(read_log, m)?)?;
    Ok(())
}