Schema Architecture¶

This page documents the internal architecture of schema handling in rypipe. It covers FrozenSchema, DiscoveryOpts, the schema cache, ensure_schema, and the fast/merge export paths.

For the adapter-author guide, see Schema.

Overview¶

Schema in rypipe answers three questions:

1. Which columns are extracted? (schema_order projection, or all discovered columns)
2. What type is each column? (field_types or default String)
3. What order are they in? (schema_order or first-appearance)

When schema_order is declared it acts as a projection + order: the output contains exactly the listed columns, in the listed order; unlisted fields are rejected by plan.resolve_field during parsing (so ColumnarSink::wants/resolve return false/None and adapters skip them entirely); listed columns absent from the data are null-filled at finish. Fields referenced by the row filter are still materialized internally so predicates evaluate correctly, but they are projected out of the final batch. A column listed in both schema_order and drop_fields is dropped (drop_fields wins).

The engine resolves these questions before parsing starts, builds an immutable FrozenSchema, and shares it across all workers. This guarantees that every batch has identical columns, types, and order.

Data flow¶

User provides schema_order + field_types
        |
        v
ExecutionPlan (compiled, shared via Arc)
        |
        +---> FrozenSchema::from_plan (explicit) -----> ensure_schema
        |                                                      |
        +---> FrozenSchema::from_discovered (auto) ----> ensure_schema
                                                               |
                                                               v
                                                    TableBuilder per chunk
                                                               |
                                                               v
                                                    sort_columns (finish)
                                                               |
                                                               v
                                                    RecordBatch export

FrozenSchema¶

FrozenSchema is the engine's immutable representation of the output column layout. It is defined in crates/rypipe-core/src/schema.rs.

Structure¶

pub struct FrozenSchema {
    /// Output column names in final order.
    names: Vec<Arc<str>>,

    /// Raw input field name -> output slot index.
    /// None = dropped by plan.
    index: FxHashMap<Box<str>, Option<u32>>,

    /// Output column types, parallel to names.
    types: Vec<FieldType>,

    /// true = full scan, false = sampled.
    exact: bool,
}

Key design decisions:

• names uses Arc<str>: shared across workers without cloning.
• index maps raw names to Option<u32>: collapses rename, drop, and lookup into a single hash probe.
• types is parallel to names: type for column i is types[i].
• exact tracks discovery method: affects error messages and cache behavior.

Construction: explicit schema¶

When schema_order is provided, FrozenSchema::from_plan builds the schema directly from the declared names:

pub fn from_plan(names: &[&str], plan: &ExecutionPlan) -> Self {
    let mut index = FxHashMap::default();
    let mut types = Vec::with_capacity(names.len());

    for (slot, &name) in names.iter().enumerate() {
        let ty = plan.column_type(name);
        types.push(ty);
        index.insert(Box::from(name), Some(slot as u32));
    }

    FrozenSchema {
        names: names.iter().map(|n| Arc::from(*n)).collect(),
        index,
        types,
        exact: true,
    }
}

Properties:

• O(n) construction where n = number of columns.
• No I/O: the schema is pure computation.
• Exact: all columns are known, no sampling uncertainty.
• exact = true: enables the "unknown field is an error" behavior.

Construction: discovered schema¶

When schema_order is empty, the engine discovers column names from the file. FrozenSchema::from_discovered applies the plan's renames and drops:

pub fn from_discovered(names_in_order: &[String], plan: &ExecutionPlan) -> Self {
    let mut index = FxHashMap::default();
    let mut out_names = Vec::new();
    let mut types = Vec::new();

    for name in names_in_order {
        let resolved = plan.resolve_field(name);
        match resolved {
            Some(resolved_name) => {
                if plan.drop_fields.contains(resolved_name) {
                    index.insert(Box::from(name.as_str()), None);
                    continue;
                }
                let slot = out_names.len() as u32;
                let ty = plan.column_type(resolved_name);
                index.insert(Box::from(name.as_str()), Some(slot));
                out_names.push(Arc::from(resolved_name));
                types.push(ty);
            }
            None => {
                index.insert(Box::from(name.as_str()), None);
            }
        }
    }

    FrozenSchema {
        names: out_names,
        index,
        types,
        exact: false,
    }
}

Properties:

• Applies renames: field_map entries are resolved during construction.
• Applies drops: dropped fields get None in the index.
• exact = false: sampled discovery may miss rare columns.
• Order: follows discovery order (file order, then document order).

Construction: partial explicit schema¶

FrozenSchema::from_partial_plan is a third mode: like from_plan, but with exact = false. Declared columns keep their specified order; fields in the file that are not listed are appended in discovery order instead of causing an error.

Resolution: the hot path¶

FrozenSchema::resolve maps a raw field name to its output slot:

#[inline]
pub fn resolve(&self, raw_name: &str) -> Option<u32> {
    self.index.get(raw_name).copied().flatten()
}

resolve is used at schema boundaries (and in tests). The per-field hot path inside TableBuilder resolves names via plan.resolve_field plus a field_index hash probe (resolve_and_slot), and after the first row of a stable layout mostly bypasses hashing entirely via the expect_slot ordinal/memcmp fast path (put_field_at). An FxHashMap probe costs ~15 cycles (one hash, one comparison).

The Option<u32> return value:

• Some(slot): field maps to output column slot
• None: field is dropped (parser should skip it)

DiscoveryOpts¶

DiscoveryOpts controls how schema discovery works for files without explicit schema.

Structure¶

pub struct DiscoveryOpts {
    /// Files smaller than this are fully scanned (cheap, exact).
    pub full_scan_threshold: u64,

    /// Number of strided windows for large files.
    pub windows: usize,

    /// Bytes per window.
    pub window_bytes: usize,

    /// Whether to always scan the tail of the file to catch
    /// late-appearing columns.
    pub always_scan_tail: bool,
}

Default values¶

impl Default for DiscoveryOpts {
    fn default() -> Self {
        Self {
            full_scan_threshold: 128 * 1024 * 1024, // 128 MiB
            windows: 16,
            window_bytes: 2 * 1024 * 1024, // 2 MiB
            always_scan_tail: true,
        }
    }
}

Discovery strategy¶

Small files (<=128 MiB):

• Full scan of the entire file
• Exact: every column is found
• Cost: one I/O pass + parsing

Large files (>128 MiB): window count and size scale with file size (dynamic_window_count / dynamic_window_size):

• 128 MiB to 1 GiB: 16 windows of 2 MiB (32 MiB total)
• 1 GiB to 10 GiB: 32 windows of 4 MiB (128 MiB total)
• 10 GiB and up: 64 windows of 4 MiB (256 MiB total)
• Windows are strided evenly through the file
• Columns found in any window are included
• Cost: ~5.3 ms for a 533 MB file on 16 threads (parallel sampling)

Accuracy¶

Sampled discovery captures columns that appear in at least one window. For typical CR XML exports:

• 100% of columns appear in the first 2 MiB (header section)
• FieldG (30% sparse) is found in ~94% of samples
• Text21 (1% sparse) is found in ~94% of samples
• A column appearing only in the last 1% of the file would be missed

When a missed column is encountered at parse time, the engine raises Error::Merge with the message:

unknown field "LateColumn" not in frozen schema (10 columns, exact=false);
pass schema=[...] with full column list or use full-scan discovery

When discovery is unreliable¶

For formats where schema discovery is unreliable (extremely sparse or heterogeneous data), pass an explicit schema from the API layer. An explicit schema pre-declares the columns: listed columns are parsed into their declared field_types in the given order, columns absent from the data come out null-filled, and unlisted fields are skipped at parse level. Fields referenced by filter are still parsed for predicate evaluation, then projected out of the output.

Schema cache¶

For batch workloads (many files with the same layout), the engine caches discovered schemas to avoid redundant discovery.

Cache structure¶

pub static SCHEMA_CACHE: LazyLock<RwLock<FxHashMap<SchemaCacheKey, SchemaCacheValue>>>
    = LazyLock::new(|| RwLock::new(FxHashMap::default()));

type SchemaCacheKey = (u64, u64);   // (file_len, sample_hash)
type SchemaCacheValue = Arc<Vec<String>>;  // discovered column names

Layout signature¶

layout_signature computes a cheap key for caching:

pub fn layout_signature(bytes: &[u8], opts: &DiscoveryOpts) -> (u64, u64) {
    let mut hasher = FxHasher::default();
    if (bytes.len() as u64) < opts.full_scan_threshold {
        // Small file: hash the entire content
        bytes.hash(&mut hasher);
    } else {
        // Large file: hash 4 strided windows
        let n = 4usize;
        let w = opts.window_bytes.min(bytes.len());
        for i in 0..n {
            let start = (bytes.len() - w) * i / n.max(1);
            bytes[start..start + w].hash(&mut hasher);
        }
    }
    (bytes.len() as u64, hasher.finish())
}

Properties:

• Fast: ~1 us for small files, ~5 us for large files.
• Collision-resistant: 4-window hash for large files.
• Stable: same layout always produces the same key.

Cache operations¶

Insert (insert_schema_cache):

pub fn insert_schema_cache(sig: (u64, u64), order: Arc<Vec<String>>) {
    let mut cache = SCHEMA_CACHE.write().unwrap_or_else(|e| e.into_inner());
    if cache.len() >= SCHEMA_CACHE_CAP {
        if let Some(oldest) = cache.keys().next().copied() {
            cache.remove(&oldest);
        }
    }
    cache.insert(sig, order);
}

• Evicts arbitrary entry when at capacity (128).
• FxHashMap does not track insertion order; this is acceptable for a bounded cache.

Lookup: done via SCHEMA_CACHE.read() in the discovery path

Clear (clear_schema_cache):

pub fn clear_schema_cache() {
    SCHEMA_CACHE.write().unwrap().clear();
    SCHEMA_CACHE_HITS.store(0, Ordering::Relaxed);
    SCHEMA_CACHE_MISSES.store(0, Ordering::Relaxed);
}

Cache statistics¶

pub static SCHEMA_CACHE_HITS: AtomicU64 = AtomicU64::new(0);
pub static SCHEMA_CACHE_MISSES: AtomicU64 = AtomicU64::new(0);

pub fn schema_cache_stats() -> (u64, u64) {
    (
        SCHEMA_CACHE_HITS.load(Ordering::Relaxed),
        SCHEMA_CACHE_MISSES.load(Ordering::Relaxed),
    )
}

For 1,000 files with the same layout:

• First file: miss (5.3 ms discovery + cache insert).
• Files 2-1000: hit (cache lookup ~1 us each).
• Total: 5.3 ms + 1 ms = 6.3 ms (vs 5.3 s without cache).

ensure_schema¶

ensure_schema is called by each worker before parsing a chunk. It guarantees that all columns from the FrozenSchema exist in the TableBuilder.

Algorithm¶

pub fn ensure_schema(&mut self, schema: &FrozenSchema) -> Result<()> {
    for (slot, name) in schema.column_names().iter().enumerate() {
        if self.field_index.contains_key(name) {
            continue; // already present
        }
        let ty = schema.column_types()[slot].clone();
        let col = ColumnBuilder::with_capacity(self.estimated_rows, &ty);
        self.columns.push(col);
        self.field_index.insert(name.to_string(), self.columns.len() - 1);
        self.column_order.push(name.to_string());
    }
    // Resize row_dirty bitmask to cover all columns.
    let words = self.columns.len().div_ceil(64);
    self.row_dirty.resize(words, 0);
    // Freeze: any later field not in this schema is an unknown field
    // and must hard-error on finish().
    self.frozen = Some(Arc::new(schema.clone()));
    Ok(())
}

Steps:

1. Iterate over all columns in the schema.
2. For each column not already in the builder, create a capacity-reserved typed column builder for the correct FieldType. No rows are pre-filled: ensure_schema runs before parsing, when the builder is empty.
3. Freeze the schema on the builder so unknown fields hard-error at finish.
4. After this call, the builder has all declared columns.

Why pre-declare the columns?¶

Consider a file where column "D" appears only in rows 500-1000. Without ensure_schema:

• Chunk 1 (rows 0-499): no column "D"
• Chunk 2 (rows 500-999): has column "D"
• At merge time: column "D" is missing from chunk 1, causing schema mismatch

With ensure_schema:

• Chunk 1: column "D" exists as an empty builder; each row that lacks it is null-filled per row by finish_row's dirty-mask pass
• Chunk 2: column "D" has values
• At export time: batches have identical schemas

Performance¶

ensure_schema is called once per chunk, not per row. For 16 chunks, it adds ~0.1 ms total (negligible vs parse time).

sort_columns¶

sort_columns reorders the internal column list to match schema_order. Note: when schema_order is non-empty, finish() takes the projection path instead: it emits exactly the declared columns in the declared order (schema columns are pre-declared in with_plan, and any still missing are created null-filled), so sort_columns is effectively vestigial for the projection path.

Algorithm¶

pub fn sort_columns(&mut self) {
    if self.plan.schema_order.is_empty() {
        return;
    }
    let order = &self.plan.schema_order;
    let rank =
        |name: &String| order.iter().position(|n| n == name).unwrap_or(usize::MAX);
    self.column_order.sort_by_key(rank);
}

Properties:

• Stable sort: columns not in schema_order (key usize::MAX) preserve relative order after the ordered ones.
• O(n log n) where n = number of columns (typically <100).
• Called once: at finish time, not per row.

Example¶

Given columns ["D", "A", "C", "B"] and schema_order = ["A", "B", "C"]:

1. ["D", "A", "C", "B"] with rank keys [usize::MAX, 0, 2, 1]
2. "D" not in schema_order -> key = usize::MAX
3. "A" at position 0 -> key = 0
4. "C" at position 2 -> key = 2
5. "B" at position 1 -> key = 1
6. Sorted (stable): ["A", "B", "C", "D"]

Fast path vs merge path¶

The engine has two export paths. The choice depends on whether all batches have identical schemas.

Fast path (parallel export)¶

Conditions:

• auto_dict is false, and all chunks agree on column variant keys (guaranteed by ensure_schema + schema_order), OR
• auto_dict is true and schemas are consistent after a parallel per-chunk auto_dict_upgrade (dictionaries are unified across chunks first when needed)

Behavior:

• Each batch is exported independently in parallel
• engines_to_record_batches builds one RecordBatch per chunk; it also runs unify_variants / promote_to_variant to reconcile storage variants across chunks before export
• No sequential merge step needed
• Throughput: ~4,980 MB/s on 533 MB

Merge path (sequential export)¶

Conditions:

• Chunk schemas are inconsistent after upgrade (or inconsistent without auto_dict): chunks disagree on a column's storage variant in a way the per-chunk export cannot reconcile ahead of time

Behavior:

• Batches are merged sequentially into one TableBuilder via extend
• merged.finish() produces a single batch
• Throughput: ~4,497 MB/s on 533 MB (-10%)

Path selection¶

Simplified from ParallelExecutor::parse:

if !plan.auto_dict && schemas_consistent(&engines) {
    return engines_to_record_batches(engines, &plan);
}
if plan.auto_dict {
    engines.par_iter_mut().for_each(|e| e.auto_dict_upgrade());
    if schemas_consistent(&engines) {
        unify_dictionaries_across_chunks(&mut engines);
        return engines_to_record_batches(engines, &plan);
    }
}
// Merge path.
let mut merged = TableBuilder::with_plan(...);
for engine in engines {
    merged.extend(engine)?;
}
Ok(vec![merged.finish()?])

schemas_consistent checks that all engines agree on column variant keys. This is guaranteed when schema_order is set.

Column type resolution¶

ExecutionPlan::column_type resolves the storage type for a column:

pub fn column_type(&self, name: &str) -> FieldType {
    // 1. Explicit type override
    if let Some(ty) = self.field_types.get(name) {
        return ty.clone();
    }

    // 2. Dictionary column
    if self.dictionary_columns.contains(name) {
        return FieldType::Dictionary;
    }

    // 3. Default: string
    FieldType::String
}

Priority:

1. field_types[name] (explicit override).
2. dictionary_columns contains name (dictionary encoding).
3. FieldType::String (default).

This is called during ensure_schema to create the correct column builder.

Unknown fields and strict types¶

The crate declares an UnknownFieldPolicy enum (Error / Skip) and re-exports it, but no code path currently reads it; it is reserved for future use. The actual behavior is fixed:

Error behavior (default)¶

With explicit schema (exact = true), unknown fields are recorded during parsing and cause a hard error at finish():

unknown field "LateColumn" not in frozen schema (10 columns, exact=true);
pass schema=[...] with full column list or use full-scan discovery

The same error with exact=false fires when sampled discovery missed a rare column (see Accuracy). This is the safe default: it prevents silent data loss. With a partial schema (from_partial_plan, exact = false), unknown fields are instead appended as new columns in discovery order.

Unknown fields surface as Error::Merge. A related parse-time error mode is strict_types: when set, a non-null string value that does not parse as the declared field_types type is recorded and surfaced at finish() as Error::Plan.

Integration with the pipeline¶

How schema flows through the engine¶

ExecutionPlan
    |
    +---> FrozenSchema::from_plan
    |         |
    |         +---> ParallelStreamOpts.schema
    |         |         |
    |         |         +---> DiscoverySink (if empty)
    |         |         |         |
    |         |         |         +---> sampled scan
    |         |         |         +---> FrozenSchema::from_discovered
    |         |         |
    |         |         +---> FrozenSchema::from_plan (if set)
    |         |
    |         +---> TableBuilder.ensure_schema (per chunk)
    |         |         |
    |         |         +---> ColumnBuilder::with_capacity (per missing column)
    |         |         +---> freeze schema on the builder
    |         |
    |         +---> sort_columns (at finish)
    |         |         |
    |         |         +---> stable-sort column_order to match schema_order
    |         |
    |         +---> engines_to_record_batches (fast path)
    |                   |
    |                   +---> one RecordBatch per chunk (parallel)
    |
    +---> TableBuilder.push_field
              |
              +---> plan.resolve_field(raw_name) + field_index probe
              |         |
              |         +---> Some(slot) -> put into column slot
              |         +---> None -> skip (dropped)
              |
              +---> expect_slot memcmp fast path (stable layout, row 2+)
              |
              +---> observer hooks (if plan.observer is set):
              |         on_begin_row / on_put_field /
              |         on_row_accepted / on_row_rejected
              |
              +---> column_type(name) -> typed builder

How schema interacts with projection¶

schema_order itself is the primary projection mechanism: unlisted fields are rejected by plan.resolve_field, so wants("unlisted_field") returns false and the parser skips the field entirely (no scanning, no decoding). The wanted-column set (schema columns plus filter-referenced fields) is pre-declared in TableBuilder::with_plan, which makes the wanted_mask/row_satisfied short-circuit stable from the first row, so adapters can byte-jump to the row close as soon as every wanted column has a value. At finish time the batch contains exactly the declared columns in the declared order; filter-only columns are projected out, and declared columns absent from the data are null-filled.

When DropFields is used, the plan's drop_fields set is populated. During schema construction:

• from_plan: the dropped field is still in the schema (caller declared it)
• from_discovered: the dropped field gets None in the index

During parsing:

• wants("dropped_field") returns false (engine checks drop_fields)
• The parser skips the field entirely (no scanning, no decoding)

At export time:

• Dropped fields are not in the output (they were never built)

How schema interacts with rename¶

When RenameFields is used, the plan's field_map is populated. During schema construction:

• from_plan: the output name is used (rename already applied by caller)
• from_discovered: resolve_field(name) applies the rename, then maps to the output slot

During parsing:

• put_field("raw_name", ...) is resolved via field_map to the output name
• The value goes into the renamed column

How schema interacts with filter¶

When FilterRows is used, the plan's filter is populated. During parsing:

• Predicate-first buffered evaluation: field values are buffered per row; the predicate is evaluated as soon as its fields are seen (eval_predicate), and at end of row the buffer is either committed to the columns (drain_buffered) or discarded without touching them.
• Adaptive fallback: when the predicate column's ordinal is >= 4/5 of the column count, buffering is disabled (buffer_worthwhile = false) and the engine falls back to direct push with pop-on-reject in finish_row (pop() per column when the predicate fails).
• Comparison flavor is per predicate: Equal/NotEqual are string-based; Compare/CompareLiteral are typed using the columns' field_types, with numeric promotion (Int64 vs Float64 compares as float).

Performance budget¶

For a 533 MB file on a Ryzen 5800X:

Without explicit schema:

Explicit schema saves ~5.3 ms per file. For 1,000 files, that is 5.3 seconds.

Thread safety¶

FrozenSchema is Clone + Send + Sync. It is built once and shared via Arc<FrozenSchema> across all workers. No locks are needed after construction.

The schema cache uses RwLock<FxHashMap> for concurrent access:

• Read lock: schema lookup (common, concurrent)
• Write lock: schema insert (rare, serialized)

Cache statistics use AtomicU64 for lock-free increment.