Fixate is an Apache Spark DataSource V2 connector that reads fixed-width (and newline-delimited) files at near-bare-metal throughput. All decompression and record parsing runs inside a native Odin shared library. Parsed rows are serialized directly into Spark's binary UnsafeRow format in off-heap memory and handed back via Java Project Panama with zero copy.

12 × 512 MiB gzip · 140-column flat layout · 6 GB compressed on disk · Spark 3.5 · 24-core / AVX-512 · 3 runs/phase

Full-work scans run ~1.5–2.2× faster than even a hand-tuned mapPartitions JVM parser; once a query pushes down column pruning, predicates, or aggregates, Fixate skips work a row-producing parser cannot — 17–56× faster vs naive, 13–43× vs optimized JVM. Full per-phase table ↓

1. Architecture Overview
2. Compression / Input Format
3. Threading Model
4. Off-Heap Memory Layout
5. DataSource V2 Class Flow
6. Supported Spark Types
7. Schema Definition & Column Mapping
8. Query Pushdowns
9. Reader Options
10. Null Handling
11. Parse Modes
12. UnsafeRow Writer Invariant
13. Build Instructions
14. Testing
15. Performance Benchmark
16. Known Limitations

%%{init: {'theme':'dark'}}%%
flowchart TD
    FS["Hadoop FileSystem\n(local / HDFS / S3 / GCS)"]:::blue
    BGReader["FixateBackgroundReader\nScala daemon thread\nreads io_buffer_size chunks"]:::green
    RB1["Ring Buffer 1\noff-heap raw input bytes\n(plaintext or gzip)\nPanama Arena"]:::blue
    NativeEngine["Native Odin Engine\nruns on Spark task thread\nvia fixate_next_rows downcall"]:::orange
    ZLIB["zlib-ng (if compression=gzip)\nstatically linked, built with AVX-512\nauto GZIP/zlib detection\nelse verbatim copy"]:::orange
    SIMD["SIMD Delimiter Scanner\nAVX-512 (64B) → AVX2 (32B) → SSE2 (16B) → scalar\nchosen at init by CPUID"]:::orange
    Parser["Field Parser\nSWAR int/long + fast-float\ndecimal i64/i128 fast path\nUnsafeRow serialization"]:::orange
    BatchBuf["4 MB Batch Row Buffer\noff-heap UnsafeRow bytes"]:::blue
    SparkTask["Spark Task Thread\nFixatePartitionReader.next()\nreusableRow.pointTo — zero copy"]:::green
    Catalyst["Catalyst Execution Engine\nWhole-Stage Code Generation"]:::green

    FS --> BGReader
    BGReader -->|"writes raw input bytes\natomic write_ptr"| RB1
    RB1 --> NativeEngine
    NativeEngine --> ZLIB
    NativeEngine --> SIMD
    NativeEngine --> Parser
    Parser --> BatchBuf
    BatchBuf -->|"(addr, size) array\nisTrivial Panama downcall"| SparkTask
    SparkTask --> Catalyst

    classDef green fill:#3b1f4f,stroke:#c084fc,stroke-width:2px,color:#f1f5f9
    classDef blue fill:#1f3b2f,stroke:#4ade80,stroke-width:2px,color:#f1f5f9
    classDef orange fill:#1e3a5f,stroke:#60a5fa,stroke-width:2px,color:#f1f5f9

• Single off-heap arena. The entire memory footprint for one partition reader — the ring buffer, decompressor scratch, row buffers, blueprint trees, and filter tables — lives in one Panama Arena.allocate(size, 64) call. The native engine never calls malloc/free during parsing.
• Batch pull API. The Spark task thread calls fixate_next_rows(engine, 2048, ptrArray, sizeArray) and receives up to MAX_BATCH_SIZE (2048) UnsafeRow pointers per downcall. The downcall is annotated Linker.Option.isTrivial() — no GC safepoint transition, no stop-the-world overhead.
• SIMD delimiter scanning. At initialization the Odin engine resolves the fastest byte-search routine for the host CPU by CPUID: AVX-512 (64B) → AVX2 (32B) → SSE2 (16B) → scalar. The AVX-512 path (simd.u8x64 compare with a 64-bit lane mask) is chosen only when avx512f + avx512bw + avx512vl are all present at runtime (avx512vl is required because the 64-lane compare legalizes through 256-bit masked compares); otherwise it falls back to AVX2 → SSE2 → scalar. Selection is purely runtime — the shared library still builds for baseline x86-64 and only executes the AVX-512 instructions on capable CPUs. A scalar tail handles the remaining bytes.
• Kernel dispatch table. The byte-scan and numeric-parse kernels (char_search, parse_long, parse_f64, parse_decimal) are resolved once at engine init by CPUID and bundled into a single Dispatch struct on the EngineInstance (populated in fixate_initialize). The per-field hot loops call straight through eng.dispatch.<kernel> — no per-call CPU-feature query or type-keyed re-selection — so a faster kernel can be slotted in behind the pointer without touching the loops. (This replaced the old single char_search_fn field.)
• SWAR integer parsing. parse_long_fast parses 8 ASCII digits per 64-bit word — mask the low nibble and fold adjacent digit groups with three multiply/shift steps, after swar_all_digits validates the word is all digits (little-endian gated; a non-digit word falls back to the scalar loop). It also takes a width-aware fast path: ≤18-digit inputs skip the per-digit overflow-guard divide (10^18 − 1 cannot overflow i64), keeping the precise guard only for 19+ digit inputs.
• Fast-float kernel. parse_f64_fast takes the Clinger fast path: accumulate the significant digits into a u64 mantissa and divide by 10^scale via an exact POW10_F64 table when digits ≤ 15 and scale ≤ 22 (both operands are exact in f64, so the single division is correctly rounded and bit-identical to strconv.parse_f64). Exponent form, too many digits, or any junk byte falls back to strconv.parse_f64. It is the default behind eng.dispatch.parse_f64.
• Bulk ptr/size array read. Per batch refill, FixatePartitionReader.next() bulk-copies the engine's off-heap row pointer/size arrays into reused JVM long[]/int[] once (MemorySegment.copy). The per-row hot path then indexes the plain Java arrays instead of issuing a MemorySegment.get per row, removing the per-row Panama scope/bounds checks while the emitted UnsafeRow stays a zero-copy view into native memory.
• zlib-ng decompression. Statically linked, compiled with AVX-512. zlib-ng runtime-dispatches AVX-512 variants relevant to the inflate/decompression path on capable CPUs — inflate_fast_avx512, chunkmemset_safe_avx512, and crc32_vpclmulqdq_avx512 (gzip's CRC32 over decompressed bytes), plus adler32_avx512 for zlib streams — gated by its own CPUID dispatch (nothing needs enabling). Initialized with windowBits = 47 (= 15 + 32) for automatic GZIP/zlib/deflate header detection. Only initialized when the input is gzip (compression=gzip).
• Plaintext fast path. The default compression=none reads uncompressed input: bytes drained from Ring Buffer 1 are copied verbatim into the decode buffer (io_thread_copy_chunk) with no inflate step, and the decompressor is never set up.
• Zero-copy inflate feed (no copy amplification). io_thread_setup_stream_inputs (fixate-core/src/parser/io_thread.odin) hands zlib-ng a direct pointer into Ring Buffer 1's contiguous span up to the ring boundary — it never stages the unread bytes through a temporary buffer. When the unread span wraps the ring, the wrapped tail is consumed on the next call once the read index wraps to 0. (The old path copied the entire unread RB1 span — up to rb1_size — into a rb1_temp_buf on every inflate step, while inflate only consumed enough input to fill its output buffer; that produced a hundreds-fold memcpy amplification that scaled with rb1_size. The rb1_temp_buf allocation has been removed entirely.) This change dropped a full-parse pass on a 12 × 1.2 GB gzip benchmark from ~410–440 s to ~138 s (~3×), and a perf profile showed libc memcpy fall from 44.3% to 0.3% of CPU — the workload is now genuinely parse-bound (~50% parsing, ~12% JVM, ~1% inflate).
• Small fixed decode buffer. The native decode/inflate-output buffer (local_buf) is a fixed 1 MB (FixateConstants.DECODE_BUFFER_SIZE), deliberately decoupled from io_buffer_size (the disk-read chunk) — the two are independent concerns. The value is still threaded to the native engine as the decode_buf_size parameter on fixate_initialize / fixate_get_required_memory_size. A perf profile showed inflate's CPU share rising monotonically with this buffer (≈3.5% at 1 MB → 4.4% at 4 MB → 5.5% at 16 MB): a larger buffer spills L2 and makes inflate's sequential output writes miss cache, while the parser saw no i-cache benefit from running longer between refills. So smaller is better, floored only by the need to comfortably exceed one record (~KB).

Fixate reads either plaintext or gzip-compressed input. The compression reader option selects the path; the default is none (plaintext) — gzip input must opt in with .option("compression", "gzip").

%%{init: {'theme':'dark'}}%%
flowchart TD
    Opt["compression option\n(default: none)"]:::green
    Validate["validateCompression()\npeek leading magic bytes"]:::green
    Conflict{"declared format\nmatches content?"}:::green
    FailFast["throw IllegalArgumentException\nclear mismatch / unsupported-format message"]:::green
    SetCfg["EngineConfig.compressed\n(byte offset 6)"]:::blue
    Drain["drain bytes from Ring Buffer 1"]:::orange
    PathQ{"compressed?"}:::orange
    Copy["io_thread_copy_chunk\nverbatim memcpy → decode buffer\n(decompressor never initialized)"]:::orange
    Inflate["io_thread_decompress_chunk\nzlib-ng inflate (windowBits 47)\n→ decode buffer"]:::orange
    Parse["record demarcation + field parsing"]:::orange

    Opt --> Validate --> Conflict
    Conflict -->|"mismatch / unsupported"| FailFast
    Conflict -->|ok| SetCfg
    SetCfg --> Drain --> PathQ
    PathQ -->|"false (plaintext)"| Copy --> Parse
    PathQ -->|"true (gzip)"| Inflate --> Parse

    classDef green fill:#3b1f4f,stroke:#c084fc,stroke-width:2px,color:#f1f5f9
    classDef blue fill:#1f3b2f,stroke:#4ade80,stroke-width:2px,color:#f1f5f9
    classDef orange fill:#1e3a5f,stroke:#60a5fa,stroke-width:2px,color:#f1f5f9

Before opening the stream, FixatePartitionReader.validateCompression() peeks the file's leading bytes and throws a clear IllegalArgumentException if the declared compression contradicts the content, instead of silently emitting garbage (gzip read as plaintext) or hitting a cryptic inflate error. It recognizes these magic numbers via detectCompression:

The engine only supports gzip/zlib, so any other recognized format is reported as unsupported. A weak 2-byte zlib heuristic (looksLikeZlib) — deflate method with a header that is a multiple of 31 — is consulted only in the compression=gzip direction, because it collides with numeric plaintext (e.g. a field starting with 80…).

%%{init: {'theme':'dark'}}%%
sequenceDiagram
    participant S  as Spark Task Thread
    participant BG as FixateBackgroundReader (daemon)
    participant RB as Ring Buffer 1 (off-heap)
    participant OE as Native Odin Engine

    S->>S: new FixatePartitionReader(ctx)
    S->>OE: fixate_initialize(blueprint, config, arena)
    S->>BG: backgroundReader.start()
    activate BG
    BG->>RB: MemorySegment.copy(chunk → rb1_data)\nwritePtr += n
    BG->>OE: fixate_notify_pushed() [sema_post]

    loop batch pull
        S->>OE: fixate_next_rows(engine, 2048, ptrs, sizes)
        OE->>RB: read input bytes [readPtr advance]
        OE->>OE: gzip → zlib-ng inflate, else copy verbatim → local_buf
        OE->>OE: SIMD scan for delimiter
        OE->>OE: parse fields → UnsafeRow in batch_row_buffer
        OE-->>S: returns row count (or -2 if stalled)
        alt stalled (RB1 empty, no EOF)
            S->>OE: fixate_wait_for_data() [sema_wait]
            BG->>RB: write more bytes
            BG->>OE: fixate_notify_pushed()
        end
        S->>S: reusableRow.pointTo(addr, size) — zero copy
    end

    BG->>OE: fixate_signal_eof()
    deactivate BG
    S->>OE: fixate_destroy(engine)
    S->>S: arena.close()

Thread 1 (FixateBackgroundReader): Opens the Hadoop FSDataInputStream and reads in io_buffer_size-byte chunks (default 4 MB). Writes into Ring Buffer 1 using MemorySegment.copy with wrap-around logic. When Ring Buffer 1 is full, parks for 100 microseconds and retries, recording stall count and total stall time in the RB1 header for telemetry. Calls fixate_notify_pushed (semaphore post) after each write. On EOF or exception, calls fixate_signal_eof.

Spark task thread: Calls fixate_next_rows in a tight loop. The native engine runs cooperatively on this thread — it drains Ring Buffer 1, decompresses, scans boundaries, parses fields, and fills the 4 MB batch buffer, then returns. If Ring Buffer 1 is empty and EOF is not yet signaled, the engine returns -2; the Scala layer then calls fixate_wait_for_data (semaphore wait) before retrying. On success, reusableRow.pointTo(null, addr, size) gives Spark a zero-copy view into the off-heap batch buffer.

%%{init: {'theme':'dark'}}%%
block-beta
    columns 3
    engine["EngineInstance\n+ EngineConfig copy"]:1
    blueprint["FieldDefinition tree\n(blueprint)"]:1
    rb1hdr["RB1 Control Header\n(capacity/write_ptr/read_ptr/eof/stalls)"]:1
    rb1data["Ring Buffer 1 Data\nrb1_size bytes\n(compressed input)"]:1
    localbuf["Local Decode Buffer\n1 MB (fixed)"]:1
    rowbuf["Row Buffer\nmax_row_size + 64 KB\n(single-row scratch)"]:1
    batchbuf["Batch Row Buffer\n4 MB\n(UnsafeRow bytes)"]:1
    sched["ParseSchedule\nflat instruction array"]:1
    filter["Filter Definitions\n(if pushed)"]:1
    trailer["Trailer Pool\n(if skip_trailer_lines > 0)"]:1
    cycles["Col Cycles Array\nu64 × field_count"]:1

%%{init: {'theme':'dark'}}%%
packet-beta
    0-7: "capacity (i64)"
    8-15: "write_ptr (i64) — Thread 1"
    16-63: "_pad1 (48 bytes)"
    64-71: "read_ptr (i64) — Odin"
    72-127: "_pad2 (56 bytes)"
    128-131: "eof (i32)"
    132-135: "_padding"
    136-143: "full_stalls_count (i64)"
    144-151: "full_stalls_time_ns (i64)"

write_ptr and read_ptr are separated by a full cache line (64 bytes of padding each) to prevent false sharing between the Scala writer thread and the Odin reader. Both are monotonically increasing byte counters; modular index is ptr % capacity.

%%{init: {'theme':'dark'}}%%
classDiagram
    direction TB

    class FixateDataSource:::green {
        +inferSchema() throws IAE
        +getTable(schema, props) FixateTable
        +supportsExternalMetadata() true
    }

    class FixateTable:::green {
        +schema StructType
        +capabilities BATCH_READ
        +newScanBuilder(options) FixateScanBuilder
    }

    class FixateScanBuilder:::green {
        -prunedSchema StructType
        -limitValue Int
        -pushedPredicatesArray Array~Predicate~
        -pushedAgg Option~FixateAggSpec~
        +pruneColumns(required)
        +pushLimit(n) Boolean
        +pushPredicates(preds) residual[]
        +pushAggregation(agg) Boolean
        +pushTableSample(...) Boolean
        +build() FixateScan
    }

    class FixateScan:::green {
        +readSchema() pruned or agg partial
        +toBatch() FixateBatch
        +supportedCustomMetrics() custom metrics
    }

    class FixateBatch:::green {
        +planInputPartitions() one per file
        +createReaderFactory() FixatePartitionReaderFactory
    }

    class FixatePartitionReaderFactory:::green {
        +createReader(partition) FixatePartitionReader
    }

    class FixatePartitionReader:::green {
        -arena Arena
        -engineHandle MemorySegment
        -reusableRow UnsafeRow
        -backgroundReader FixateBackgroundReader
        +next() Boolean
        +get() InternalRow
        +close()
        +currentMetricsValues()
    }

    class FixateBackgroundReader:::green {
        -readerThread Thread daemon
        +start()
        +stop()
        +getAndResetBytesRead() Long
    }

    class NativeOdinEngine:::orange {
        fixate_initialize()
        fixate_next_rows()
        fixate_set_aggregates()
        fixate_destroy()
    }

    FixateDataSource --> FixateTable
    FixateTable --> FixateScanBuilder
    FixateScanBuilder --> FixateScan
    FixateScan --> FixateBatch
    FixateBatch --> FixatePartitionReaderFactory
    FixatePartitionReaderFactory --> FixatePartitionReader
    FixatePartitionReader --> FixateBackgroundReader
    FixatePartitionReader --> NativeOdinEngine

    classDef green fill:#3b1f4f,stroke:#c084fc,stroke-width:2px,color:#f1f5f9
    classDef orange fill:#1e3a5f,stroke:#60a5fa,stroke-width:2px,color:#f1f5f9

FixateSchemaCompiler translates each Spark DataType into a native FieldDefinition. Any type not in the table below throws UnsupportedOperationException at scan-initialization time.

Boolean parsing rules. A field is true if (after whitespace trimming):

• The trimmed value exactly matches the string in schema metadata key true_value.
• Or it is true / TRUE (case-insensitive, 4 characters).
• Or it is one of the single characters T, t, Y, y, or 1.

DateType / TimestampType are parsed by a general pattern interpreter (native fixdatetime) driven by the column's format metadata — any well-formed fixed-width pattern works, not a fixed allow-list:

• Date/time tokens: yyyy/uuuu (4-digit year), yy/uu (2-digit → 20yy), MM/M, dd/d, HH/H, mm/m, ss/s, SSS… (fractional seconds). Any other character is a literal that must match (-, /, :, , .), and '…' quotes a literal section (e.g. the 'T' in yyyy-MM-dd'T'HH:mm:ss). Examples: yyyyMMdd, yyyyMM (day ⇒ 1), uuuuMMdd, dd-MM-yyyy, yyyyMMddHHmmss, yyyy-MM-dd HH:mm:ss.SSS.
• Year token semantics: yyyy (year-of-era) requires a positive year and rejects year 0 (1 BCE). The proleptic uuuu token accepts year 0. Pathological/overflowing year values yield null rather than error.
• Calendar validation: months and per-month day counts are validated with correct leap-year rules — 2023-02-29 (non-leap) and 2021-04-31 are rejected (null in PERMISSIVE, error in FAILFAST).
• Timezone (Timestamp only). The instant is resolved as: (1) an offset token in the value — X (+05/Z), XX (+0530), XXX (+05:30) — read per row (DST-correct); else (2) the column's timezone metadata (a fixed offset); else (3) UTC. DST-bearing IANA zones in timezone metadata are rejected — use an offset token for per-row DST correctness. DateType is timezone-independent.
• No format ⇒ the field is read as an integer epoch value (epoch-days for Date, epoch-micros for Timestamp).

Fixate requires an explicit schema. FixateDataSource.inferSchema always throws IllegalArgumentException. Supply the schema via .schema(...).

import org.apache.spark.sql.types._

val schema = new StructType()
  .add("id",     LongType,   nullable = true,
       new MetadataBuilder().putLong("offset", 0).putLong("length", 10).build())
  .add("name",   StringType, nullable = true,
       new MetadataBuilder().putLong("offset", 10).putLong("length", 30).build())
  .add("amount", DecimalType(15, 4), nullable = true,
       new MetadataBuilder().putLong("offset", 40).putLong("length", 15).build())

spark.read
  .format("io.codeberg.hectormiguel.spark.fixate.FixateDataSource")
  .schema(schema)
  .option("record_length", "55")
  .option("compression", "gzip")   // omit (or "none") for plaintext input
  .load("/data/input/*.gz")

Field lengths can also be supplied as reader options when modifying the schema is inconvenient:

spark.read
  .format("io.codeberg.hectormiguel.spark.fixate.FixateDataSource")
  .schema(schema)
  .option("col.id.length", "10")
  .option("col.name.length", "30")
  .load(path)

When both metadata and col.<name>.length are present, metadata takes precedence.

%%{init: {'theme':'dark'}}%%
flowchart TD
    Query["Spark query plan"]:::green
    PruneQ{"Column pruning\npossible?"}:::green
    Prune["pruneColumns()\nOnly requested columns\ncompiled into blueprint"]:::green
    LimitQ{"LIMIT\npushed?"}:::green
    Limit["pushLimit(n)\nEngine stops after n records\nSpark re-checks as safety net"]:::green
    FilterQ{"Filter on top-level column\n(EqualTo/LT/LTE/GT/GTE/IsNull/IsNotNull/In/StringStartsWith\nfor supported types)?"}:::green
    Filter["pushPredicates()\nAND/OR/NOT trees +\ncomparison ops on String/Int/Long/Short/Byte/Date/Float/Double/Decimal\nIsNull + IsNotNull on any top-level column\nIn on Int/Long/Short/Byte/String\nStringStartsWith on String\n(Boolean: EqualTo only)\nTimestamp / nested → Spark"]:::green
    FilterFB["Unsupported filters\nreturned to Spark"]:::green
    AggQ{"Global aggregate\nno GROUP BY?"}:::green
    Agg["pushAggregation()\nCOUNT★, COUNT(col), SUM, MIN, MAX\nApplies any pushed filters, then accumulates\none partial row/partition; Spark merges"]:::green
    AggFB["Aggregation stays\nin Spark"]:::green
    Scan["FixateScan\n(build)"]:::green

    Query --> PruneQ
    PruneQ -->|yes| Prune
    PruneQ -->|no| Scan
    Prune --> LimitQ
    LimitQ -->|yes| Limit
    LimitQ -->|no| FilterQ
    Limit --> FilterQ
    FilterQ -->|supported| Filter
    FilterQ -->|unsupported| FilterFB
    Filter --> AggQ
    FilterFB --> AggQ
    AggQ -->|"accepted\n(COUNT★/COUNT(col)/SUM/MIN/MAX\nover supported types)"| Agg
    AggQ -->|"declined\n(GROUP BY or unsupported func/type)"| AggFB
    Agg --> Scan
    AggFB --> Scan

    classDef green fill:#3b1f4f,stroke:#c084fc,stroke-width:2px,color:#f1f5f9

FixateScanBuilder implements five DSv2 pushdown interfaces (SupportsPushDownRequiredColumns, SupportsPushDownLimit, SupportsPushDownV2Filters, SupportsPushDownAggregates, SupportsPushDownTableSample):

A filter on a top-level column is pushed to the native engine only for the types below; the byte position is resolved against the original schema, so a filter on a column pruned from the output still locates its bytes. Date filters carry the column's format (epoch-day comparison); a Date column with no format compares as integer epoch days. Everything else falls back to post-scan evaluation in Spark.

Additional constraints applied before pushing any filter:

• Out-of-range literals: an Int/Short/Byte literal that falls outside the column type's range is NOT pushed — Spark evaluates post-scan with the correct semantics.
• -0.0 normalization: a -0.0 Float/Double literal is normalized to +0.0 (adding 0.0) so the pushed comparand's bits match Spark's -0.0 == +0.0 semantics.
• NaN / ±Infinity: non-finite Float/Double literals are never pushed; the native comparator uses raw IEEE compares (NaN != NaN) which diverges from Spark.
• Record bounds: a filter whose source window (offset + length) exceeds the declared record_length is not pushed.

Pushed aggregates are global only (no GROUP BY). COUNT(*) works regardless of column type. Date/Timestamp MIN/MAX compare as epoch days / epoch micros; Boolean compares as false < true. SUM is numeric-only (Spark disallows SUM on date/timestamp/boolean) — and Decimal SUM is intentionally declined (see the note below).

Why Decimal SUM is not pushed. Spark 3.5's V2 aggregate-pushdown rewrite always casts the pushed partial column back to the original Sum.child type (Decimal(p,s)) before re-summing. A per-partition partial that exceeds precision p overflows to NULL at that cast. No choice of partial type survives: declaring the partial as Decimal(p,s) causes the native partial value itself to overflow; declaring it as Decimal(p+10,s) causes Spark's forced CAST(partial AS Decimal(p,s)) to overflow. Since Spark always narrows back to the source precision, FixateAggregateCompiler declines Decimal SUM entirely and lets Spark compute it with its own lossless widening to Decimal(p+10,s). MIN/MAX/COUNT over Decimal still push normally (their partials are actual element values that fit within Decimal(p,s)).

If any aggregate in the query references an unsupported function (e.g. distinct), or a type not listed for that aggregate, the entire aggregation is declined and Spark performs it.

All option names support both snake_case (canonical) and camelCase (legacy, deprecated). The canonical snake_case form is preferred.

Custom metrics are exposed per partition in the Spark UI:

For best CPU utilization, set spark.task.cpus = 2 so the Scala I/O daemon and the Spark task thread do not compete for a single core.

Four independent null mechanisms can be combined:

1. Blank-as-null (blank_as_null = true, default) — A field that is entirely whitespace after positional slicing is treated as null. This is the primary mechanism for space-padded fixed-width files.
2. Null guard sentinel (null_guard) — A specific string that, when matched exactly after trimming, marks any field null. Applied globally to all columns.
3. Per-column null sentinel (null_value field metadata) — A sentinel string scoped to a single column. When the trimmed field exactly matches the column's null_value, that column is null. This composes with the global null_guard.
4. Zero-as-null (zero_numerics_as_null = false, default) — When enabled, any numeric field whose parsed value is 0 is treated as null.

Pushed COUNT(col) / SUM / MIN / MAX apply the same null definition as the full-parse path: global blank_as_null / null_guard, the per-column null_value sentinel (carried into the native FixateAggDef), and zero_numerics_as_null. So an aggregate ignores exactly the rows Spark would see as null. Specifically:

• COUNT(*) counts every DATA record (including all-null rows) and is never null.
• COUNT(col) counts only non-null values; the result is never null (0 if none).
• SUM / MIN / MAX ignore nulls and return NULL (not 0) when there were zero non-null values.

Each partition emits one partial-aggregate row; Spark performs the final cross-partition merge (supportCompletePushDown = false).

A record is considered malformed if:

• In fixed-length mode (record_length > 0): the record byte count does not equal record_length.
• A field fails a bounds check (parsed integer outside the target type's range).
• A decimal parse fails entirely.

When a fixed-width record fails bounds validation (its byte count does not equal record_length), PERMISSIVE nulls ALL non-corrupt fields and routes the raw bytes to _corrupt_record — it does NOT attempt to salvage leading fields that appear to be present. In fixed-width framing, a record that fails bounds validation cannot have trusted field boundaries: a "present" leading field may be a truncation artifact. This differs deliberately from Spark CSV PERMISSIVE, which salvages parsed tokens up to the parse failure. Only the per-field parse-failure path (a field whose value fails to parse, e.g. a non-numeric integer) preserves successfully-parsed sibling fields; the short/oversize-record path null-alls. (Verified in ParseModeEdgeSpec Gap#3 test.)

Records that exceed the native decode buffer (~1 MiB, FixateConstants.DECODE_BUFFER_SIZE) are treated as malformed and honor the configured parse mode rather than always raising a hard error. Under DROPMALFORMED, the oversize record is silently dropped and parsing continues. Under PERMISSIVE, the buffered prefix is routed to _corrupt_record and the remainder is drained. Under FAILFAST, the engine flags an error and the task fails. (Verified in ParseModeEdgeSpec E2/BUG-004 tests.)

The native UnsafeRow writers are built around a hard invariant that contributors must preserve:

Every field in a row is written exactly once — a value OR a null, never both.

row_writer_init zeroes the entire null bitset once up front (and nested-array writers do the same via write_nested_array_init). Because the bitset starts all-zero, the typed row_writer_write_* procedures no longer clear the per-field null bit — only row_writer_write_null ever sets a bit. This removed a per-field read-modify-write on a hot cache line (measurable at 100+ columns/row) and the write-after-write hazard it created between adjacent same-word field writes.

The consequence for record builders: null only the fields a record type does NOT populate; never "null all fields then overwrite". Pre-nulling a field that is later written a value would leave its null bit set after the value write, silently nulling a populated column. (This was a real bug in the header/trailer row builders — write_header_row / write_trailer_row now skip the fields they populate and null only the rest.)

A secondary parser optimization: parse_long_fast skips the per-digit overflow-guard divide for inputs of ≤18 digits, since 10^18 − 1 < i64 max cannot overflow; the precise per-digit guard is kept only for 19+ digit inputs.

• Odin compiler (LLVM-based, latest stable)
• CMake (to build zlib-ng)
• Java 21 (OpenJDK 21; --enable-preview is required)
• Maven 3.x
• Optional cross-compile toolchains:
  • aarch64-linux-gnu-gcc for Linux ARM64
  • x86_64-w64-mingw32-gcc for Windows x64
  • x86_64-apple-darwin-clang / aarch64-apple-darwin-clang for macOS

# 1. Build native library for the host platform.
#    Cross-compiles other targets if their toolchains are present.
#    Copies all artifacts under fixate-spark/src/main/resources/natives/<platform>/
make build-native

# 2. Compile the Scala connector (requires step 1)
make build-scala

# 3. Both together
make build

# Debug build (bounds checking enabled, no optimizations)
make -C fixate-core libfixate_core.so DEBUG=1

# Clean everything
make clean

All Linux/Windows artifacts are cross-compiled from one x86_64 host (make build-native); zlib-ng is statically linked into each (no system libz dependency at runtime). macOS needs an Apple toolchain not available via apt.

The Linux x86_64 native library is compiled with -microarch:x86-64-v3, which sets x86-64-v3 as the baseline ISA (Haswell 2013+, AMD Excavator 2015+). This enables BMI/BMI2 (tzcnt, pdep, pext), FMA, and unaligned SSE/AVX access in scalar code. Any cloud or EMR instance type in active use today meets this baseline — only pre-2013 x86 hardware (before Intel Haswell / AMD Excavator) would not. The ARM64 and Windows cross-compiled libraries do not inherit this flag and set their own architecture-appropriate defaults.

SIMD paths (AVX-512 delimiter scan, AVX2, SSE2) are still selected at runtime by CPUID, so a Haswell CPU without AVX-512 uses the AVX2 path without error — the -microarch:x86-64-v3 flag only affects the scalar code generation baseline.

scripts/perf_analyze.py is a streamlined native perf-data analysis script for the Fixate engine. It processes a perf.data file (and optional perf stat output) and emits a structured Markdown report covering per-event self-time filtered to the native .so, call-graph top callers, and instruction-level hotspots for chosen symbols. See the script's docstring for usage.

The Panama FFI and Spark internals require these flags (already set in the Makefile MAVEN_OPTS for test and benchmark targets):

--enable-preview
--enable-native-access=ALL-UNNAMED
--add-opens=java.base/java.lang=ALL-UNNAMED
--add-opens=java.base/java.lang.invoke=ALL-UNNAMED
--add-opens=java.base/java.lang.reflect=ALL-UNNAMED
--add-opens=java.base/java.io=ALL-UNNAMED
--add-opens=java.base/java.net=ALL-UNNAMED
--add-opens=java.base/java.nio=ALL-UNNAMED
--add-opens=java.base/java.util=ALL-UNNAMED
--add-opens=java.base/java.util.concurrent=ALL-UNNAMED
--add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED
--add-opens=java.base/sun.nio.ch=ALL-UNNAMED
--add-opens=java.base/sun.nio.cs=ALL-UNNAMED
--add-opens=java.base/sun.security.action=ALL-UNNAMED
--add-opens=java.base/java.security=ALL-UNNAMED
--add-opens=java.base/sun.util.calendar=ALL-UNNAMED

The sun.util.calendar open is required by Spark's date/time conversion path, which is exercised when DateType / TimestampType columns are produced.

# Native Odin unit tests (SIMD scanner, ring buffer, decimal serialization)
make test-native

# Spark integration tests
make test-scala

# Both
make test

The suite is 271 tests: 190 Scala/Spark integration specs (src/test/scala) and 81 native Odin unit tests (*_test.odin), all passing.

The native suite is one set of tests gated by when ODIN_ARCH + runtime CPU-feature checks (not per-arch suites): the cross-platform paths (scalar, 128-bit SIMD, parsing/decimal/datetime/aggregate) run everywhere; the x86 AVX2/AVX-512 paths run only on amd64, and AVX-512 only when the CPU advertises it. CI executes the suite on the x86_64 build host, so it covers the x86 SIMD + shared paths; ARM64 is build-verified (cross-compiled), not executed — its NEON path runs only on an arm64 runner.

CI runs on a self-hosted Forgejo Actions runner (Codeberg) using a prebuilt image (ci/Containerfile) that bakes the whole toolchain — JDK 21, Maven, Node, CMake, Odin, and the mingw-w64 / aarch64 cross toolchains — plus a pre-warmed Maven repo, so runs do no per-run setup. Each push:

1. runs the native (Odin) + Scala (Spark) test suites,
2. builds the engine and cross-compiles the Windows x64 + Linux ARM64 libraries,
3. packages the multi-arch JAR and uploads it as a build artifact.

The zlib-ng build is cached on the runner keyed by the submodule commit (rebuilt only when the submodule is bumped). On pushes to master, the JAR is additionally published as a dated release (tag build-<UTC timestamp>), one per master build.

The benchmark (PerformanceBenchmark) compares Fixate against two independently-toggled JVM baselines:

• Baseline 1 — naive substring .map. Reads each file as text and parses columns with substring + trim + .toInt / .toBigDecimal. Toggle with -Dbenchmark.run.baseline (default true).
• Baseline 2 — optimized hand-rolled JVM parser. Uses .mapPartitions with a reused Row and scratch buffer, index slicing instead of substring, and a fall-through integer parse. Toggle with -Dbenchmark.run.baseline2 (default false).

The two flags are independent — you can run either, both, or neither.

# 0.5 GB per file, 5 files, nested layout, with Baseline 1 (naive) comparison
make benchmark BENCHMARK_SIZE=0.5 BENCHMARK_FILES=5 BENCHMARK_BASELINE=true

make benchmark wires BENCHMARK_BASELINE to -Dbenchmark.run.baseline. To enable Baseline 2 (or other system properties), pass them via MAVEN_OPTS / the exec:java invocation directly.

12 files × 512 MiB gzip (6.00 GB compressed on disk) · 140-column flat layout · 3 runs/phase · 24-core box · AVX-512 native build · Spark 3.5.0. The flat layout exercises every flat parse path (var-len String, all inline numerics, small-decimal i64 and large-decimal i128/BigInteger, format-driven and integer-epoch Date/Timestamp). Baseline 1 = naive substring .map; Baseline 2 = optimized hand-rolled .mapPartitions JVM parser. Both baselines parse all 140 columns (the fair comparison).

Reading the table. The JVM-side optimizations are real — Baseline 2 is ~1.4× faster than the naive substring baseline. Yet on the full-work scan (Phase 1) Fixate is still 2.08× faster than naive and 1.46× faster than the hand-tuned JVM parser on raw parse alone. The dramatic phases (2, 4–10) are where a query pushes column pruning, predicates, or aggregates into the native engine: a row-producing JVM parser must materialize every column of every row regardless, while Fixate skips that work entirely — 17–56× vs naive, 13–43× vs the optimized JVM.

These are flat-layout numbers (the scalar-parse story across all 14 type groups). The nested struct/array layout was not part of this measurement run.

Eliminating the per-step copy of the entire unread Ring-Buffer-1 span into a temp buffer (see Zero-copy inflate feed above) was the largest single performance win to date. On a 12 × 1.2 GB gzip dataset, a full-parse pass dropped from ~410–440 s to ~138 s (~3×), and a perf profile showed libc memcpy fall from 44.3% to 0.3% of CPU. The workload is now genuinely parse-bound (~50% parsing, ~12% JVM, ~1% inflate), so further gains come from the field-parsing kernels rather than the I/O path.

Benchmark data files are stored under benchmark_data/<layout>/benchmark_N.gz and regenerated only when missing.

• DateType / TimestampType parse a general format pattern (see Date & timestamp parsing) with full calendar/leap-year validation. With no format, the field is read as an integer epoch value (epoch-days for Date, epoch-micros for Timestamp). Timestamp timezone is taken from an offset token in the value, else the timezone metadata, else UTC; DST-bearing IANA zones in timezone metadata are rejected (a fixed column offset can't represent DST — use an offset token in the format instead). The proleptic year-of-era token uuuu accepts year 0 (1 BCE); the calendar year-of-era token yyyy rejects year 0. Pathological / overflowing year values yield null rather than error.
• Filter pushdown is limited to top-level scalar columns. Comparison ops (EqualTo / LessThan / LessThanOrEqual / GreaterThan / GreaterThanOrEqual) are pushed for StringType, IntegerType, LongType, ShortType, ByteType, DateType, FloatType, DoubleType, and DecimalType; EqualTo only for BooleanType. IsNull / IsNotNull are pushed for any top-level column. In is pushed for Int/Long/Short/Byte/String. StringStartsWith is pushed for String. Timestamp comparison filters are deliberately left to Spark pending timezone-parity verification. Filters on nested StructType / ArrayType columns, out-of-range literals, and filter windows exceeding record_length are not pushed. NaN and ±Infinity Float/Double literals are not pushed.
• Decimal SUM is not pushed. SUM over Decimal(p,s) is declined because Spark 3.5's V2 SUM rewrite casts the pushed partial column back to Decimal(p,s) before re-summing, causing any partition partial that exceeds precision p to overflow to NULL. Spark computes decimal SUM itself with lossless widening. MIN / MAX / COUNT over Decimal still push.
• PERMISSIVE fixed-width null-all. A short/malformed fixed-width record (byte count ≠ record_length) under PERMISSIVE nulls ALL fields and routes raw bytes to _corrupt_record — field boundaries in a truncated record are unreliable. This differs from Spark CSV PERMISSIVE, which salvages parsed tokens up to the failure point.
• Aggregate pushdown is partial only. Each partition emits one partial-aggregate row and Spark performs the final cross-partition merge; supportCompletePushDown returns false. (Pushed filters and a pushed aggregate do compose — filtered rows are excluded from the accumulation.)
• No schema inference. A schema must be supplied via .schema(...).
• One partition per file. Each matched file maps to exactly one InputPartition. There is no intra-file splitting.
• Fixed-width ArrayType only. Arrays are read as a fixed-count slice of the record. Variable-length or delimited arrays are not supported.
• Trailing trailer records dropped by LIMIT. When a LIMIT is pushed and the engine stops early, the trailing skip_trailer_lines records may not be buffered yet; they are silently dropped. This is a known limitation of the cooperative row-count limit mechanism.