The ColumnarSink Trait¶

ColumnarSink is the bridge between your parser and the engine. The parser calls begin_row/put_field/end_row for each record; the sink accumulates values into typed Arrow columns.

See Decoder API for how the engine implements this trait internally.

Method reference¶

Required¶

Field resolution¶

Tier control¶

Projection¶

Layout prediction¶

Batch¶

The default just loops over put_field, so there is nothing to gain unless you override it. Override only if your parser naturally produces a whole row at once (a decoded record struct, a fixed-width slot array) and can push it without per-field branching; otherwise the per-field methods give the engine more chances to short-circuit (wants, row_rejected).

Fast paths in order¶

The engine provides four push methods, from fastest to slowest. The speed differences come from skipped lookups, and each skip moves a responsibility onto your parser: the faster the method, the more preconditions you must guarantee yourself. The slower methods exist because they are simpler and robust by construction; a few nanoseconds per field only matter on very hot paths, so start with put_field and move down this list only where profiling justifies it.

1. put_field_at(slot, value): fastest¶

Direct slot push. No name resolution, no hash lookup. Used by the expect_slot path after the layout is learned.

expect_slot(ordinal) → Some((slot, expected))
  memcmp(raw, expected) == 0  →  put_field_at(slot, value)

Cost: ~5 ns per field (column write + dirty bit set).

Use when: your format has a fixed field order and the parser sits on a hot path (millions of rows).

Constraint: slot numbers are only meaningful for one specific layout. You must guard every push with the expect_slot memcmp and call layout_broken(ordinal) the moment a row deviates (reordered, missing, or extra fields), or you silently write values into the wrong columns. See Layout prediction.

2. put_field_resolved(name, value): fast¶

Skips the rename lookup. Used when you've already called resolve().

resolve(name) → Some(resolved)
  put_field_resolved(resolved, value)

Cost: ~10 ns per field (single HashMap lookup + column write).

Use when: extraction is expensive (entity decoding, base64, date parsing), so you want to check resolve() once, skip unwanted fields, and push without paying the rename/drop lookup a second time.

Constraint: the resolved name is only valid within the current row state; do not cache it across rows or schema changes. And if extraction is cheap, this gains nothing over resolve_and_put, since you paid for resolve() anyway.

3. resolve_and_put(name, value): medium¶

Single resolve + push. Default implementation.

resolve(name) → Some(resolved)
  put_field_resolved(resolved, value)

Cost: ~15 ns per field (HashMap lookup + column write).

Use when: you want the resolve() semantics (rename + drop honored, unwanted fields skipped) in one call without managing the resolved name yourself. This is the default for a reason: it is what put_field degenerates to after resolution.

Constraint: still one hash lookup per field; on a fixed-layout hot loop, method 1 or 2 avoids it.

4. put_field(name, value): slowest¶

Full resolve + push. The engine calls resolve_field(name) which checks rename map, then drop set, then returns the output name.

Cost: ~20 ns per field (two HashMap lookups + column write).

Use when: almost everywhere. No bookkeeping, no layout assumptions, no validity windows: pass the raw field name and the value and the engine does the right thing. At ~20 ns per field, a 1M-row, 10-column file spends about 0.2 s of total parse time here, usually a small fraction of actual parsing work.

Constraint: none, which is the point.

The projection fast path¶

When a projection selects 3 of 11 columns and all 3 arrive by field 4, the scanner can byte-jump to the row close tag, skipping fields 5-11.

sink.row_satisfied()  →  true  →  scanner calls find_row_close()

This is implemented in the scanner as:

// After each child element:
if sink.row_satisfied() {
    let after = find_row_close(bytes, cur, row_tag, regions);
    sink.end_row();
    return Flow::At(after);
}

Composes with row_rejected() (filter rejection). A row can be satisfied OR rejected; both short-circuit the scanner.

wanted_mask() provides the bitmask for this:

fn wanted_mask(&self) -> u64 {
    // Bitmask of columns in the output schema
    let mut mask = 0u64;
    for name in &self.plan.schema_order {
        if let Some(&idx) = self.field_index.get(name) {
            if idx < 64 { mask |= 1u64 << idx; }
        }
    }
    mask
}

The adapter checks (mask >> slot) & 1 == 1 instead of calling wants() per field.

The layout prediction fast path¶

After the first row, the engine knows which slot each ordinal maps to. expect_slot(ordinal) returns (slot, raw_name_bytes). The adapter compares raw bytes via memcmp instead of running the full attribute scan → UTF-8 decode → hash → lookup pipeline.

expect_slot(ordinal)  →  Some((slot, expected))
  memcmp(raw, expected) == 0  →  put_field_at(slot, value)  // skip decode + resolve
  memcmp(raw, expected) != 0  →  layout_broken(ordinal)     // invalidate, fall back

Cost comparison:

• Generic path: ~25 ns (find_attr_value + decode_attr + resolve + put_field)
• Fast path: ~8 ns (memcmp + put_field_at)

When it helps: Formats with stable field order across rows (CSV columns, XML attributes, JSONL keys). The first row learns the layout; subsequent rows skip the expensive resolution.

When it doesn't help: Formats where field order changes between rows, or where field names contain entities that need decoding before comparison.

The predicate-first fast path¶

When a filter is active, the engine buffers fields until the predicate resolves. If the predicate passes mid-row, the engine switches to direct mode and drains the buffer.

begin_row → [put_field × N] → end_row
                │
                ▼
         check predicate slot
         ├── Pass → direct mode (push remaining fields directly)
         ├── Fail → discard buffer, skip row
         └── Undecided → continue buffering

The scanner checks sink.row_rejected() after each child element:

if sink.row_rejected() {
    let after = find_row_close(bytes, cur, row_tag, regions);
    sink.end_row();
    return Flow::At(after);
}

Adaptive strategy: If the predicate column appears late (> 4/5 of columns), buffering is a net loss. The engine switches to direct push + pop-on-reject.

Example: minimal sink¶

struct CountingSink {
    rows: usize,
    fields: usize,
}

impl ColumnarSink for CountingSink {
    fn begin_row(&mut self) {}
    fn put_field(&mut self, _name: &str, _value: Value<'_>) { self.fields += 1; }
    fn end_row(&mut self) { self.rows += 1; }
    fn finish(&mut self) -> Result<RecordBatch> {
        Ok(RecordBatch::new_empty(Arc::new(Schema::empty())))
    }
}

Example: profiling sink (locate-only)¶

struct LocateOnlySink {
    row_count: usize,
    field_count: usize,
    plan: ExecutionPlan,
}

impl ColumnarSink for LocateOnlySink {
    fn begin_row(&mut self) {}
    fn put_field(&mut self, name: &str, _value: Value<'_>) {
        self.field_count += 1;
        // Only resolve, don't store
        let _ = self.plan.resolve_field(name);
    }
    fn end_row(&mut self) { self.row_count += 1; }
    fn wants(&self, _name: &str) -> bool { true }
    fn needs_value(&self) -> bool { false }  // Skip text extraction
    fn resolve<'a>(&'a self, name: &'a str) -> Option<&'a str> {
        self.plan.resolve_field(name)
    }
    fn finish(&mut self) -> Result<RecordBatch> {
        Ok(RecordBatch::new_empty(Arc::new(Schema::empty())))
    }
}

Thread safety¶

ColumnarSink is Send but not Sync. Each chunk gets its own sink instance via begin_row/end_row lifecycle. The engine creates one TableBuilder per chunk and merges them after all chunks complete.

Build and test¶

Verify the projection contract: with a plan that drops every field except name, a correct wants() implementation lets the parser skip all other fields and the finished batch has exactly one column:

$ cargo test sink_wants
running 1 test
test tests::sink_wants_skips_dropped_fields ... ok

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

The test builds a plan with .drop("age").drop("active"), parses a two-row sample, and asserts batch.num_columns() == 1 and the remaining column is name. If your sink ignores wants(), all three columns appear and this test fails.