The Splitter Trait¶

The Splitter trait decides where it is safe to divide an input byte stream into independent chunks. The engine calls find_split_points to get byte offsets, then parses each chunk concurrently via rayon.

See Architecture for how the engine uses split points internally.

Trait definition¶

pub trait Splitter: Send + Sync {
    /// The only required method: the next record boundary at or after `from`.
    /// Must return a position where a record starts, or None.
    fn next_record_start(&self, bytes: &[u8], from: usize) -> Option<usize>;

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

    /// Optional: byte ranges where a candidate boundary must be rejected
    /// (comments, CDATA, quoted fields, string literals). See Skip regions.
    fn skip_regions(&self) -> Option<&dyn SkipRegionFinder> { None }

    /// Provided. Do not override without a measured reason.
    fn find_split_points(&self, bytes: &[u8], max_chunks: usize) -> Vec<usize>;
}

Required: next_record_start¶

This is the only method you must implement. Given a byte position, return the position where the next record starts at or after that position.

fn next_record_start(&self, bytes: &[u8], from: usize) -> Option<usize> {
    // For newline-delimited formats:
    memchr::memchr(b'\n', &bytes[from..])
        .map(|rel| from + rel + 1)
}

Rules:

• Return Some(position) where position is the first byte of the next record.
• Return None if no more records exist after from.
• The position must be valid: position <= bytes.len().
• Do not return a position at a delimiter; return the position of the first byte of the record itself.

Why this works: The engine calls next_record_start at nominal offsets (bytes.len() * i / n) to find the nearest record boundary. Your implementation just needs to answer "where does the next record start from here?" The engine handles deduplication, sorting, and chunk planning.

Required: estimate_bytes_per_row¶

Called once on a sample of the input (first 64 KB) to estimate row size. The bounded executor uses this to plan chunk sizes. Simple newline-counting suffices for most formats:

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

Optional: skip_regions¶

If your format has regions where a candidate delimiter must be ignored (comments, CDATA, quoted fields, string literals), implement skip_regions():

fn skip_regions(&self) -> Option<&dyn SkipRegionFinder> {
    Some(&MySkipRegions)
}

See Skip regions for the full SkipRegionFinder interface and implementation examples.

Default: find_split_points¶

Do not override this method unless you have a measured reason. The default implementation provides:

1. Nominal offsets at bytes.len() * i / n for i in 1..n
2. Serial boundary search over nominal offsets, each calling next_record_start; parsing remains parallel after ranges are built
3. Skip-region rejection via in_skip_region (bounded backward scan)
4. Dedup and sort
5. Chunk sizing via plan_chunk_count (2 MiB target, thread minimums, 1024-chunk cap)

fn find_split_points(&self, bytes: &[u8], max_chunks: usize) -> Vec<usize> {
    let n = plan_chunk_count(bytes.len(), max_chunks, SplitMode::Parallel);
    let nominals: Vec<usize> = (1..n).map(|i| bytes.len() * i / n).collect();

// serial scan over nominals, each calling next_record_start
    // reject candidates inside skip regions
    // dedup, sort, prepend 0, append bytes.len()
}

The default is strictly better than hand-rolled splitting because it applies the measured chunk-size target (MIN_CHUNK_BYTES = 2 MiB) that guides the sub-1 MB chunk collapse. See Chunk planning.

What the engine does with split points¶

1. find_split_points returns vec![0, 13, 26, 39, ..., bytes.len()]
2. The engine converts these to ranges: [0..13, 13..26, 26..39, ...]
3. Each range is parsed independently by parse_chunk on a rayon thread
4. Results are merged into a single RecordBatch (or kept chunked for streaming)

The engine guarantees:

• Every chunk contains whole records (no mid-record splits)
• Empty chunks are discarded
• Chunks are parsed in parallel with no shared mutable state

Performance characteristics¶

• next_record_start is called once per nominal offset (typically 100-200 times)
• Each call scans forward from the nominal to find the next record boundary
• The scan is O(chunk_size / row_size) on average
• Skip-region rejection adds O(window × num_openers) per candidate
• Total split time is < 1% of single-threaded parse time on 500 MB

Stateful formats¶

Multi-line formats (properties, INI, LDAP) require the Splitter to skip continued newlines and comment lines. Instead of scanning bytes manually, use the declarative methods:

use rypipe_core::{RecordBoundary, Splitter, find_next_record_boundary};

struct PropertiesSplitter;

impl Splitter for PropertiesSplitter {
    fn next_record_start(&self, bytes: &[u8], from: usize) -> Option<usize> {
        find_next_record_boundary(
            bytes, from,
            self.continuation_char(),
            self.comment_prefixes(),
            self.record_boundary() == RecordBoundary::BlankLine,
        )
    }

    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)
    }

    // Declare format rules; the engine handles the scanning
    fn record_boundary(&self) -> RecordBoundary { RecordBoundary::Line }
    fn continuation_char(&self) -> Option<u8> { Some(b'\\') }
    fn comment_prefixes(&self) -> &[&[u8]] { &[b"#", b"!"] }
}

find_next_record_boundary handles:

• Continuations: skips \n preceded by the continuation character
• Comments: skips lines starting with any comment prefix
• Blank lines: when record_boundary = BlankLine, splits on \n\n

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: track [section] headers, accumulate fields across lines, etc. The state persists across begin_row/end_row calls within one parse_chunk call.

If a state spans a chunk boundary (e.g., a \ continuation that crosses from one chunk to the next), the Splitter must prevent that split; the declarative methods handle this automatically.

Limitations¶

Formats where the parser's output affects the next chunk's parsing (e.g., "row N changes how row N+1 is interpreted") cannot be parallelized. These require sequential processing, which the engine does not currently offer as a first-class mode. Use engine="columnar" (single chunk) for such formats.

Common mistakes¶

1. Overriding find_split_points: Bypasses chunk sizing and skip-region rejection. The default is almost always better.

2. Splitting inside records: Each chunk must contain whole records. Split at record boundaries, not at arbitrary byte offsets.

3. Not handling empty input: find_split_points returns vec![0, 0] for empty input. Your next_record_start should return None for empty input.

4. Returning positions at delimiters: Split points should be at the first byte of a record, not at the delimiter itself. For CSV, return the byte after \n, not the \n itself.

5. Ignoring skip regions: If your format has comments or quoted fields, implement skip_regions. Without it, the engine may split inside a comment or quoted string, producing corrupt chunks.

6. Using find_split_points for row-level iteration: find_split_points is for chunk-level splitting only. Row-level iteration uses parse_chunk.

Build and test¶

Unit-test the Splitter against a fixed sample: next_record_start must return the byte after each \n, and estimate_bytes_per_row must divide the sample by its newline count:

$ cargo test splitter
running 2 tests
test tests::splitter_estimates_bytes_per_row ... ok
test tests::splitter_finds_row_starts ... ok

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

The test asserts next_record_start(sample, 0) == Some(30) (first row ends at byte 29), next_record_start(sample, 30) == Some(sample.len()), and None past the end. A wrong position here corrupts every chunk boundary, so run this before benchmarking anything.

What the end user sees¶

None of this machinery is visible to the person using your adapter. Once a correct Splitter is registered, the engine picks split points and parallel threads on its own:

from rypipe_log import LogSource

# The splitter runs behind the scenes; parallelism is automatic.
table = LogSource("sample.log").to_arrow()